|
5 | 5 | allowing fine-grained control over which roles can access which storage protocols. |
6 | 6 | """ |
7 | 7 |
|
| 8 | +import json |
8 | 9 | import os |
9 | 10 | import sys |
10 | 11 |
|
11 | 12 | sys.path.insert(1, os.path.join(sys.path[0], "../..")) |
12 | 13 |
|
13 | 14 | import pytest |
14 | 15 |
|
| 16 | +from opteryx.config import RESOURCES_PATH |
15 | 17 | from opteryx.managers.permissions import can_read_table |
16 | 18 |
|
| 19 | + |
| 20 | +@pytest.fixture(scope="module", autouse=True) |
| 21 | +def setup_test_permissions(): |
| 22 | + """ |
| 23 | + Set up test permissions for protocol prefix tests. |
| 24 | + These permissions define roles with different levels of access to protocols. |
| 25 | + """ |
| 26 | + # Define test permissions |
| 27 | + test_permissions = [ |
| 28 | + {"role": "restricted", "permission": "READ", "table": "opteryx.*"}, |
| 29 | + {"role": "data_analyst", "permission": "READ", "table": "opteryx.*"}, |
| 30 | + {"role": "data_analyst", "permission": "READ", "table": "gs://*"}, |
| 31 | + {"role": "data_engineer", "permission": "READ", "table": "opteryx.*"}, |
| 32 | + {"role": "data_engineer", "permission": "READ", "table": "file://*"}, |
| 33 | + {"role": "data_engineer", "permission": "READ", "table": "gs://*"}, |
| 34 | + {"role": "data_engineer", "permission": "READ", "table": "s3://*"}, |
| 35 | + {"role": "cloud_only", "permission": "READ", "table": "gs://*"}, |
| 36 | + {"role": "cloud_only", "permission": "READ", "table": "s3://*"}, |
| 37 | + {"role": "project_team", "permission": "READ", "table": "gs://project-bucket/*"}, |
| 38 | + ] |
| 39 | + |
| 40 | + # Backup original permissions file |
| 41 | + permissions_file = RESOURCES_PATH / "permissions.json" |
| 42 | + backup_file = RESOURCES_PATH / "permissions.json.bak" |
| 43 | + |
| 44 | + original_content = None |
| 45 | + if permissions_file.exists(): |
| 46 | + with open(permissions_file, "r") as f: |
| 47 | + original_content = f.read() |
| 48 | + |
| 49 | + # Write test permissions |
| 50 | + with open(permissions_file, "w") as f: |
| 51 | + for perm in test_permissions: |
| 52 | + f.write(json.dumps(perm) + "\n") |
| 53 | + |
| 54 | + # Reload permissions module |
| 55 | + from opteryx.managers import permissions as perm_module |
| 56 | + perm_module.PERMISSIONS = perm_module.load_permissions() |
| 57 | + |
| 58 | + # Run tests |
| 59 | + yield |
| 60 | + |
| 61 | + # Restore original permissions |
| 62 | + if original_content is not None: |
| 63 | + with open(permissions_file, "w") as f: |
| 64 | + f.write(original_content) |
| 65 | + elif backup_file.exists(): |
| 66 | + backup_file.unlink() |
| 67 | + |
| 68 | + # Reload original permissions |
| 69 | + from opteryx.managers import permissions as perm_module |
| 70 | + perm_module.PERMISSIONS = perm_module.load_permissions() |
| 71 | + |
17 | 72 | # Test cases for protocol prefix permissions treated as table namespaces |
18 | 73 | test_cases = [ |
19 | 74 | # Basic protocol namespace matching |
|
0 commit comments