|
| 1 | +from os import getenv |
| 2 | +import pytest |
| 3 | +from dj_gui_api_server.DJGUIAPIServer import app |
| 4 | +import datajoint as dj |
| 5 | +from base64 import b64encode |
| 6 | +from json import dumps |
| 7 | + |
| 8 | + |
| 9 | +@pytest.fixture |
| 10 | +def client(): |
| 11 | + with app.test_client() as client: |
| 12 | + yield client |
| 13 | + |
| 14 | + |
| 15 | +@pytest.fixture |
| 16 | +def token(client): |
| 17 | + yield client.post('/api/login', json=dict(databaseAddress=getenv('TEST_DB_SERVER'), |
| 18 | + username=getenv('TEST_DB_USER'), |
| 19 | + password=getenv('TEST_DB_PASS'))).json['jwt'] |
| 20 | + |
| 21 | + |
| 22 | +@pytest.fixture |
| 23 | +def connection(): |
| 24 | + dj.config['safemode'] = False |
| 25 | + connection = dj.conn(host=getenv('TEST_DB_SERVER'), |
| 26 | + user=getenv('TEST_DB_USER'), |
| 27 | + password=getenv('TEST_DB_PASS'), reset=True) |
| 28 | + connection.query(""" |
| 29 | + CREATE USER IF NOT EXISTS 'underprivileged'@'%%' |
| 30 | + IDENTIFIED BY 'datajoint'; |
| 31 | + """) |
| 32 | + connection.query("GRANT ALL PRIVILEGES ON `deps`.* TO 'underprivileged'@'%%';") |
| 33 | + deps_secret = dj.VirtualModule('deps_secret', 'deps_secret', create_tables=True) |
| 34 | + deps = dj.VirtualModule('deps', 'deps', create_tables=True) |
| 35 | + @deps.schema |
| 36 | + class TableA(dj.Lookup): |
| 37 | + definition = """ |
| 38 | + a_id: int |
| 39 | + --- |
| 40 | + a_name: varchar(30) |
| 41 | + """ |
| 42 | + contents = [(0, 'Raphael',), (1, 'Bernie',)] |
| 43 | + |
| 44 | + @deps.schema |
| 45 | + class TableB(dj.Lookup): |
| 46 | + definition = """ |
| 47 | + -> TableA |
| 48 | + b_id: int |
| 49 | + --- |
| 50 | + b_number: float |
| 51 | + """ |
| 52 | + contents = [(0, 10, 22.12), (0, 11, -1.21,), (1, 21, 7.77,)] |
| 53 | + deps = dj.VirtualModule('deps', 'deps', create_tables=True) |
| 54 | + |
| 55 | + @deps_secret.schema |
| 56 | + class DiffTableB(dj.Lookup): |
| 57 | + definition = """ |
| 58 | + -> deps.TableA |
| 59 | + bs_id: int |
| 60 | + --- |
| 61 | + bs_number: float |
| 62 | + """ |
| 63 | + contents = [(0, -10, -99.99), (0, -11, 287.11,)] |
| 64 | + |
| 65 | + @deps.schema |
| 66 | + class TableC(dj.Lookup): |
| 67 | + definition = """ |
| 68 | + -> TableB |
| 69 | + c_id: int |
| 70 | + --- |
| 71 | + c_int: int |
| 72 | + """ |
| 73 | + contents = [(0, 10, 100, -8), (0, 11, 200, -9,), (0, 11, 300, -7,)] |
| 74 | + |
| 75 | + yield connection |
| 76 | + |
| 77 | + deps_secret.schema.drop() |
| 78 | + deps.schema.drop() |
| 79 | + connection.query("DROP USER 'underprivileged'@'%%';") |
| 80 | + connection.close() |
| 81 | + dj.config['safemode'] = True |
| 82 | + |
| 83 | + |
| 84 | +@pytest.fixture |
| 85 | +def underprivileged_token(client, connection): |
| 86 | + yield client.post('/api/login', json=dict(databaseAddress=getenv('TEST_DB_SERVER'), |
| 87 | + username='underprivileged', |
| 88 | + password='datajoint')).json['jwt'] |
| 89 | + |
| 90 | + |
| 91 | +def test_dependencies_underprivileged(underprivileged_token, client): |
| 92 | + schema_name = 'deps' |
| 93 | + table_name = 'TableA' |
| 94 | + restriction = b64encode(dumps(dict(a_id=0)).encode('utf-8')).decode('utf-8') |
| 95 | + REST_dependencies = client.get( |
| 96 | + f"""/api/record/dependency?schemaName={ |
| 97 | + schema_name}&tableName={table_name}&restriction={restriction}""", |
| 98 | + headers=dict(Authorization=f'Bearer {underprivileged_token}')).json['dependencies'] |
| 99 | + REST_records = client.post('/api/fetch_tuples', |
| 100 | + headers=dict(Authorization=f'Bearer {underprivileged_token}'), |
| 101 | + json=dict(schemaName=schema_name, |
| 102 | + tableName=table_name)).json['tuples'] |
| 103 | + assert len(REST_records) == 2 |
| 104 | + assert len(REST_dependencies) == 4 |
| 105 | + table_a = [el for el in REST_dependencies |
| 106 | + if el['schema'] == 'deps' and 'table_a' in el['table']][0] |
| 107 | + assert table_a['accessible'] and table_a['count'] == 1 |
| 108 | + table_b = [el for el in REST_dependencies |
| 109 | + if el['schema'] == 'deps' and 'table_b' in el['table']][0] |
| 110 | + assert table_b['accessible'] and table_b['count'] == 2 |
| 111 | + table_c = [el for el in REST_dependencies |
| 112 | + if el['schema'] == 'deps' and 'table_c' in el['table']][0] |
| 113 | + assert table_c['accessible'] and table_c['count'] == 3 |
| 114 | + diff_table_b = [el for el in REST_dependencies |
| 115 | + if el['schema'] == 'deps_secret' and 'diff_table_b' in el['table']][0] |
| 116 | + assert not diff_table_b['accessible'] |
| 117 | + |
| 118 | + |
| 119 | +def test_dependencies_admin(token, client, connection): |
| 120 | + schema_name = 'deps' |
| 121 | + table_name = 'TableA' |
| 122 | + restriction = b64encode(dumps(dict(a_id=0)).encode('utf-8')).decode('utf-8') |
| 123 | + REST_dependencies = client.get( |
| 124 | + f"""/api/record/dependency?schemaName={ |
| 125 | + schema_name}&tableName={table_name}&restriction={restriction}""", |
| 126 | + headers=dict(Authorization=f'Bearer {token}')).json['dependencies'] |
| 127 | + REST_records = client.post('/api/fetch_tuples', |
| 128 | + headers=dict(Authorization=f'Bearer {token}'), |
| 129 | + json=dict(schemaName=schema_name, |
| 130 | + tableName=table_name)).json['tuples'] |
| 131 | + assert len(REST_records) == 2 |
| 132 | + assert len(REST_dependencies) == 4 |
| 133 | + table_a = [el for el in REST_dependencies |
| 134 | + if el['schema'] == 'deps' and 'table_a' in el['table']][0] |
| 135 | + assert table_a['accessible'] and table_a['count'] == 1 |
| 136 | + table_b = [el for el in REST_dependencies |
| 137 | + if el['schema'] == 'deps' and 'table_b' in el['table']][0] |
| 138 | + assert table_b['accessible'] and table_b['count'] == 2 |
| 139 | + table_c = [el for el in REST_dependencies |
| 140 | + if el['schema'] == 'deps' and 'table_c' in el['table']][0] |
| 141 | + assert table_c['accessible'] and table_c['count'] == 3 |
| 142 | + diff_table_b = [el for el in REST_dependencies |
| 143 | + if el['schema'] == 'deps_secret' and 'diff_table_b' in el['table']][0] |
| 144 | + assert diff_table_b['accessible'] and diff_table_b['count'] == 2 |
0 commit comments