# Scenario step implementations ## Authentication setup IMPLEMENTS GIVEN an RSA key pair for token signing argv = [ os.path.join(srcdir, 'generate-rsa-key'), 'token.key', ] cliapp.runcmd(argv, stdout=None, stderr=None) IMPLEMENTS GIVEN an access token for scopes (.+) scopes = get_next_match() key = open('token.key').read() argv = [ os.path.join(srcdir, 'create-token'), scopes, ] token = cliapp.runcmd(argv, feed_stdin=key) write('token.jwt', token) vars['issuer'] = 'localhost' vars['audience'] = 'localhost' ## Controller configuration IMPLEMENTS GIVEN controller config uses (\S+) at the state directory vars['statedir'] = get_next_match() ## Start and stop the controller IMPLEMENTS GIVEN a running ick controller import os, time, cliapp, yaml vars['controller.log'] = 'ick_controller.log' vars['gunicorn3.log'] = 'gunicorn3.log' vars['port'] = random_free_port() vars['url'] = 'http://127.0.0.1:{}'.format(vars['port']) config = { 'token-issuer': vars['issuer'], 'token-audience': vars['audience'], 'token-public-key': cat('token.key.pub'), 'log': [ { 'filename': vars['controller.log'], }, ], 'statedir': vars['statedir'], } env = dict(os.environ) env['ICK_CONTROLLER_CONFIG'] = 'ick_controller.yaml' yaml.safe_dump(config, open('ick_controller.yaml', 'w')) argv = [ 'gunicorn3', '--daemon', '--bind', '127.0.0.1:{}'.format(vars['port']), '--log-file', vars['gunicorn3.log'], '--log-level', 'debug', '-p', 'pid', 'ick_controller:app', ] cliapp.runcmd(argv, env=env) vars['pid'] = int(cat('pid')) wait_for_port(vars['port']) IMPLEMENTS WHEN user stops ick controller import os, signal os.kill(int(vars['pid']), signal.SIGTERM) IMPLEMENTS FINALLY stop ick controller import os, signal os.kill(vars['pid'], signal.SIGTERM) ## HTTP requests of various kinds IMPLEMENTS WHEN user makes request GET (\S+) path = get_next_match() token = cat('token.jwt') url = vars['url'] status, content_type, body = get(url + path, token) vars['status_code'] = status vars['content_type'] = content_type vars['body'] = body IMPLEMENTS WHEN user makes request POST (\S+) (.+) path = get_next_match() body_text = get_next_match() print('path', path) print('body', body_text) token = cat('token.jwt') url = vars['url'] status, content_type, body = post(url + path, body_text, token) vars['status_code'] = status vars['content_type'] = content_type vars['body'] = body IMPLEMENTS WHEN user makes request PUT (\S+) (.+) path = get_next_match() body_text = get_next_match() print('path', path) print('body', body_text) token = cat('token.jwt') url = vars['url'] status, content_type, body = put(url + path, body_text, token) vars['status_code'] = status vars['content_type'] = content_type vars['body'] = body IMPLEMENTS WHEN user makes request DELETE (\S+) path = get_next_match() token = cat('token.jwt') url = vars['url'] status, content_type, body = delete(url + path, token) vars['status_code'] = status vars['content_type'] = content_type vars['body'] = body ## HTTP response inspection IMPLEMENTS THEN result has status code (\d+) print(cat('token.jwt')) expected = int(get_next_match()) assertEqual(expected, vars['status_code']) IMPLEMENTS THEN body matches (.+) expected_text = get_next_match() expected = json.loads(expected_text) print('actual body', repr(vars['body'])) actual = json.loads(vars['body']) assertEqual(expected, actual) IMPLEMENTS THEN version in body matches version from setup.py body = vars['body'] obj = json.loads(body) actual = obj['version'] setup_py = os.path.join(srcdir, 'setup.py') wanted = cliapp.runcmd(['python3', setup_py, '--version']).strip() assertEqual(actual, wanted) ## Controller state inspection IMPLEMENTS THEN controller state directory contains project (\S+) name = get_next_match() filename = os.path.join(vars['statedir'], 'projects', name + '.yaml') assertTrue(os.path.exists(filename))