| # Check that the LNT REST JSON API is working. |
| # create temporary instance |
| # RUN: rm -rf %t.instance |
| # RUN: python %{shared_inputs}/create_temp_instance.py \ |
| # RUN: %s %{shared_inputs}/SmallInstance \ |
| # RUN: %t.instance %S/Inputs/V4Pages_extra_records.sql |
| # |
| # RUN: python %s %t.instance %{shared_inputs} |
| |
| import json |
| import logging |
| import sys |
| import unittest |
| |
| import lnt.server.db.migrate |
| import lnt.server.ui.app |
| from V4Pages import check_json |
| |
| logging.basicConfig(level=logging.INFO) |
| |
| |
| class _hashabledict(dict): |
| """See https://stackoverflow.com/questions/1151658.""" |
| def __hash__(self): |
| return hash(tuple(sorted(self.items()))) |
| |
| |
| class JSONAPIDeleteTester(unittest.TestCase): |
| """Test the REST api.""" |
| |
| def setUp(self): |
| """Bind to the LNT test instance.""" |
| _, instance_path, shared_inputs = sys.argv |
| app = lnt.server.ui.app.App.create_standalone(instance_path) |
| app.testing = True |
| self.client = app.test_client() |
| self.shared_inputs = shared_inputs |
| |
| def test_00_update_machine(self): |
| """Check PUT request to /machines/n""" |
| client = self.client |
| |
| # We are going to set the 'os' field to none, remove the 'uname' |
| # parameter and add the 'new_parameter' parameter. |
| # Make sure none of those things happened yet: |
| machine_before = check_json(client, 'api/db_default/v4/nts/machines/1') |
| machine_before = machine_before['machine'] |
| self.assertIsNotNone(machine_before.get('os', None)) |
| self.assertIsNone(machine_before.get('new_parameter', None)) |
| self.assertIsNotNone(machine_before.get('uname', None)) |
| |
| data = { |
| 'machine': { |
| 'hardware': 'hal 9000', |
| 'os': None, |
| 'hostname': 'localhost', |
| 'new_parameter': True, |
| }, |
| } |
| json_data = json.dumps(data) |
| resp = client.put('api/db_default/v4/nts/machines/1', data=json_data, |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 200) |
| |
| machine_after = check_json(client, 'api/db_default/v4/nts/machines/1') |
| machine_after = machine_after['machine'] |
| for key in ('hardware', 'os', 'hostname', 'new_parameter', 'uname'): |
| self.assertEqual(machine_after.get(key, None), |
| data['machine'].get(key, None)) |
| |
| def test_00_rename_machine(self): |
| """Check rename POST request to /machines/n""" |
| client = self.client |
| |
| # Make sure the environment is as expected. |
| j = check_json(client, 'api/db_default/v4/nts/machines/1') |
| self.assertEqual(j['machine']['name'], 'localhost__clang_DEV__x86_64') |
| |
| data = { |
| 'action': 'rename', |
| 'name': 'new_machine_name', |
| } |
| resp = client.post('api/db_default/v4/nts/machines/1', data=data) |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.post('api/db_default/v4/nts/machines/1', data=data, |
| headers={'AuthToken': 'wrong token'}) |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.post('api/db_default/v4/nts/machines/1', data=data, |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 200) |
| |
| # Machine should be renamed now. |
| j = check_json(client, 'api/db_default/v4/nts/machines/1') |
| self.assertEqual(j['machine']['name'], 'new_machine_name') |
| |
| def test_01_delete_run(self): |
| """Check /runs/n can be deleted.""" |
| client = self.client |
| |
| j = check_json(client, 'api/db_default/v4/nts/runs/1') |
| sample_ids = [s['id'] for s in j['tests']] |
| self.assertNotEqual(len(sample_ids), 0) |
| for sid in sample_ids: |
| resp = client.get('api/db_default/v4/nts/samples/{}'.format(sid)) |
| self.assertEqual(resp.status_code, 200) |
| |
| resp = client.delete('api/db_default/v4/nts/runs/1') |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.delete('api/db_default/v4/nts/runs/1', |
| headers={'AuthToken': 'wrong token'}) |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.delete('api/db_default/v4/nts/runs/1', |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 200) |
| |
| resp = client.get('api/db_default/v4/nts/runs/1') |
| self.assertEqual(resp.status_code, 404) |
| |
| for sid in sample_ids: |
| resp = client.get('api/db_default/v4/nts/samples/{}'.format(sid)) |
| self.assertEqual(resp.status_code, 404) |
| |
| def test_02_delete_machine(self): |
| """Check /machines/n can be deleted.""" |
| client = self.client |
| |
| j = check_json(client, 'api/db_default/v4/nts/machines/2') |
| run_ids = [s['id'] for s in j['runs']] |
| self.assertNotEqual(len(run_ids), 0) |
| sample_ids = [] |
| for run_id in run_ids: |
| resp = check_json(client, |
| 'api/db_default/v4/nts/runs/{}'.format(run_id)) |
| sample_ids.append([s['id'] for s in resp['tests']]) |
| self.assertNotEqual(len(sample_ids), 0) |
| |
| resp = client.delete('api/db_default/v4/nts/machines/2') |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.delete('api/db_default/v4/nts/machines/2', |
| headers={'AuthToken': 'wrong token'}) |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.delete('api/db_default/v4/nts/machines/2', |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 200) |
| self.assertEqual(resp.get_data(as_text=True), |
| '''Deleting runs 3 5 6 7 8 9 (6/6) |
| Deleted machine machine2:2 |
| ''') |
| |
| resp = client.get('api/db_default/v4/nts/machines/2') |
| self.assertEqual(resp.status_code, 404) |
| |
| for run_id in run_ids: |
| resp = client.get('api/db_default/v4/nts/runs/{}'.format(run_id)) |
| self.assertEqual(resp.status_code, 404) |
| |
| for sid in sample_ids: |
| resp = client.get('api/db_default/v4/nts/samples/{}'.format(sid)) |
| self.assertEqual(resp.status_code, 404) |
| |
| def test_03_post_run(self): |
| """Check POST to /runs.""" |
| client = self.client |
| |
| resp = client.get('api/db_default/v4/nts/runs/5') |
| self.assertEqual(resp.status_code, 404) |
| |
| data = open('%s/sample-report.json' % self.shared_inputs).read() |
| |
| resp = client.post('api/db_default/v4/nts/runs', data=data) |
| self.assertEqual(resp.status_code, 401) |
| |
| resp = client.post('api/db_default/v4/nts/runs', data=data, |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 301) |
| self.assertIn('http://localhost/api/db_default/v4/nts/runs/', resp.headers['Location']) |
| resp_json = json.loads(resp.data) |
| self.assertEqual(resp_json['run_id'], 5) |
| |
| # Provoke a failing submission. |
| resp = client.post('api/db_default/v4/nts/runs?merge=reject', |
| data=data, |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 400) |
| resp_json = json.loads(resp.data) |
| self.assertEqual(resp_json['error'], |
| "import failure: Duplicate submission for '1'") |
| self.assertEqual(resp_json['success'], False) |
| |
| def test_04_merge_into(self): |
| """Check POST/merge into request for /machines.""" |
| client = self.client |
| |
| # Download existing machines. |
| machine_1 = check_json(client, 'api/db_default/v4/nts/machines/1') |
| machine_3 = check_json(client, 'api/db_default/v4/nts/machines/3') |
| # The test is boring if we don't have at least 1 run in each machine. |
| self.assertTrue(len(machine_1['runs']) > 0) |
| self.assertTrue(len(machine_3['runs']) > 0) |
| |
| data = { |
| 'action': 'merge', |
| 'into': '3', |
| } |
| resp = client.post('api/db_default/v4/nts/machines/1', data=data, |
| headers={'AuthToken': 'test_token'}) |
| self.assertEqual(resp.status_code, 200) |
| |
| # Old machine should have disappeared. |
| resp_2 = client.get('api/db_default/v4/nts/machines/1') |
| self.assertEqual(resp_2.status_code, 404) |
| |
| # The other machine should have the union of all runs. |
| machine_1['runs'] = [_hashabledict(run) for run in machine_1['runs']] |
| machine_3['runs'] = [_hashabledict(run) for run in machine_3['runs']] |
| allruns = set(machine_1['runs']).union(machine_3['runs']) |
| resp_3 = check_json(client, 'api/db_default/v4/nts/machines/3') |
| resp_3['runs'] = [_hashabledict(run) for run in resp_3['runs']] |
| self.assertEqual(set(resp_3['runs']), allruns) |
| |
| |
| if __name__ == '__main__': |
| unittest.main(argv=[sys.argv[0], ]) |
