c4c322e6d6
Some checks failed
Codespell / Check for spelling errors (push) Successful in 41s
Codespell / Check for spelling errors (pull_request) Successful in 41s
Ruff / Code linting (push) Successful in 52s
Ruff / Code linting (pull_request) Successful in 51s
Type annotation (PR) / check (pull_request) Failing after 1m13s
Test execution / Test-all (push) Successful in 1m26s
Add a test that ensures that instance-state is updated when a collection is `PUT` or `DELETE`-ed.
376 lines
13 KiB
Python
376 lines
13 KiB
Python
from pathlib import (
|
|
Path,
|
|
PurePosixPath,
|
|
)
|
|
|
|
from starlette.testclient import TestClient
|
|
|
|
from dump_things_service import (
|
|
HTTP_200_OK,
|
|
HTTP_201_CREATED,
|
|
HTTP_401_UNAUTHORIZED,
|
|
HTTP_404_NOT_FOUND,
|
|
)
|
|
from dump_things_service.abstract_config import (
|
|
ForgejoAuthSpec,
|
|
GitAuditBackendConfig,
|
|
TokenCollectionConfig,
|
|
TokenModes,
|
|
hash_token_representation,
|
|
)
|
|
from dump_things_service.collection_endpoints import CollectionRequest
|
|
from dump_things_service.instance_state import get_instance_state
|
|
from dump_things_service.token_endpoints import (
|
|
AdminTokenRequest,
|
|
TokenRequest,
|
|
)
|
|
|
|
# String representation of curated- and incoming-path
|
|
curated = 'admin_test_curated'
|
|
incoming = 'admin_test_incoming'
|
|
|
|
# Path to a local simple test schema
|
|
test_schema_location = str((Path(__file__).parent / 'testschema.yaml').absolute())
|
|
|
|
new_collection_name = 'admin_test_collection'
|
|
new_token_name = 'admin_test_token'
|
|
new_token_representation = 'admin_test_token'
|
|
new_collection_request = CollectionRequest(
|
|
name=new_collection_name,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/admin_test_collection'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/admin_test_collection'),
|
|
)
|
|
|
|
new_token_request = TokenRequest(
|
|
name=new_token_name,
|
|
user_id='admin_test_token_user',
|
|
hashed=False,
|
|
representation=new_token_representation,
|
|
collections={
|
|
new_collection_name: TokenCollectionConfig(
|
|
mode=TokenModes.WRITE_COLLECTION,
|
|
incoming_label=f'{new_collection_name}_label',
|
|
)
|
|
},
|
|
)
|
|
|
|
new_admin_token_name = 'New_Admin_Token'
|
|
plain_new_admin_token = 'admin-XXX'
|
|
new_admin_token_request = AdminTokenRequest(
|
|
name=new_admin_token_name,
|
|
representation=hash_token_representation(plain_new_admin_token),
|
|
)
|
|
|
|
|
|
def _name_in_openapi_paths(
|
|
test_client: TestClient,
|
|
name: str,
|
|
) -> bool:
|
|
response = test_client.get('/openapi.json')
|
|
open_api = response.json()
|
|
return any(name in path for path in open_api['paths'])
|
|
|
|
|
|
def test_collection_adding(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
# Check that the collection does not yet exist
|
|
response = test_client.get(
|
|
f'/collections/{new_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_404_NOT_FOUND
|
|
assert not _name_in_openapi_paths(test_client, new_collection_name)
|
|
|
|
# Add a new collection to the configuration
|
|
response = test_client.post(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=new_collection_request.model_dump(mode='json', by_alias=True),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Check that the collection endpoints exist in the open API document
|
|
assert _name_in_openapi_paths(test_client, new_collection_name)
|
|
|
|
# Read back the new collection and compare it
|
|
response = test_client.get(
|
|
f'/collections/{new_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
new_collection_config = new_collection_request.model_dump(
|
|
mode='json', by_alias=True
|
|
)
|
|
del new_collection_config['name']
|
|
assert response.json() == new_collection_config
|
|
|
|
# Add a token to the collection
|
|
response = test_client.post(
|
|
'/tokens',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=new_token_request.model_dump(mode='json'),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Read the token back
|
|
response = test_client.get(
|
|
f'/tokens/{new_token_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
assert response.json() == {
|
|
'name': new_token_request.name,
|
|
'user_id': new_token_request.user_id,
|
|
'collections': new_token_request.model_dump(mode='json')['collections'],
|
|
'hashed': new_token_request.hashed,
|
|
'representation': new_token_request.representation,
|
|
}
|
|
|
|
# Add a new record to the collection to verify that the token works
|
|
new_record = {
|
|
'pid': 'http://example.com/admin-test-1',
|
|
'given_name': 'Admin Test 1',
|
|
'schema_type': 'abc:Person',
|
|
}
|
|
response = test_client.post(
|
|
f'/{new_collection_name}/record/Person',
|
|
headers={'x-dumpthings-token': new_token_representation},
|
|
json=new_record,
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Read the record back
|
|
response = test_client.get(
|
|
f'/{new_collection_name}/records/Person',
|
|
headers={'x-dumpthings-token': new_token_representation},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
assert response.json()[0] == new_record
|
|
|
|
# Remove the token
|
|
response = test_client.delete(
|
|
f'/tokens/{new_token_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Check that posting is not possible with the removed token
|
|
response = test_client.post(
|
|
f'/{new_collection_name}/record/Person',
|
|
headers={'x-dumpthings-token': new_token_representation},
|
|
json=new_record,
|
|
)
|
|
assert response.status_code == HTTP_401_UNAUTHORIZED
|
|
|
|
# Remove the collection
|
|
response = test_client.delete(
|
|
f'/collections/{new_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Check that the collection endpoints are not found
|
|
response = test_client.get(
|
|
f'/{new_collection_name}/records/Person',
|
|
headers={'x-dumpthings-token': new_token_representation},
|
|
)
|
|
assert response.status_code == HTTP_404_NOT_FOUND
|
|
|
|
# Check that the openapi document is adjusted
|
|
assert not _name_in_openapi_paths(test_client, new_collection_name)
|
|
|
|
|
|
def test_collection_putting(fastapi_client_simple, tmp_path):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
put_collection_name = 'test_put_collection'
|
|
put_collection_request_orig = CollectionRequest(
|
|
name=put_collection_name,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/put_test_collection'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/put_test_collection'),
|
|
)
|
|
put_collection_request_updated = CollectionRequest(
|
|
name=put_collection_name,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/put_test_collection'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/put_test_collection_updated'),
|
|
audit_backends=[
|
|
GitAuditBackendConfig(
|
|
type='gitaudit',
|
|
path=Path(tmp_path),
|
|
auto_flush_timeout=2,
|
|
)
|
|
],
|
|
)
|
|
|
|
# Check that the collection does not yet exist
|
|
response = test_client.get(
|
|
f'/collections/{put_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_404_NOT_FOUND
|
|
assert not _name_in_openapi_paths(test_client, put_collection_name)
|
|
|
|
# Add the first version of the collection
|
|
response = test_client.post(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=put_collection_request_orig.model_dump(mode='json', by_alias=True),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
assert _name_in_openapi_paths(test_client, put_collection_name)
|
|
|
|
audit_files = tuple(tmp_path.iterdir())
|
|
assert len(audit_files) == 0
|
|
|
|
response = test_client.get(
|
|
f'/collections/{put_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Update the collection
|
|
response = test_client.put(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=put_collection_request_updated.model_dump(mode='json', by_alias=True),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Check that the audit backend is activated
|
|
audit_files = tuple(tmp_path.iterdir())
|
|
assert len(audit_files) > 0
|
|
|
|
# Delete the collection again because we check for a known number of
|
|
# collections in other tests.
|
|
response = test_client.delete(
|
|
f'/collections/{put_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
|
|
def test_collection_reading(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
# Check that the new admin token is not yet working
|
|
response = test_client.get(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
response_object = response.json()
|
|
assert isinstance(response_object, list)
|
|
assert len(response_object) == 10
|
|
|
|
|
|
def test_admin_token_management(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
# Check that the new admin token is not yet working
|
|
response = test_client.get(
|
|
'/collections/collection_1',
|
|
headers={'x-dumpthings-token': plain_new_admin_token},
|
|
)
|
|
assert response.status_code == HTTP_401_UNAUTHORIZED
|
|
|
|
# Add a new admin token
|
|
response = test_client.post(
|
|
'/admin_tokens',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=new_admin_token_request.model_dump(mode='json'),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
|
|
# Try the new token
|
|
response = test_client.get(
|
|
'/collections/collection_1',
|
|
headers={'x-dumpthings-token': plain_new_admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
# Check that the token shows up in the token list
|
|
response = test_client.get(
|
|
'/admin_tokens',
|
|
headers={'x-dumpthings-token': plain_new_admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
names = [entry['name'] for entry in response.json()]
|
|
assert new_admin_token_name in names
|
|
|
|
# Delete the new admin token
|
|
response = test_client.delete(
|
|
f'/admin_tokens/{new_admin_token_name}',
|
|
headers={'x-dumpthings-token': plain_new_admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
|
|
response = test_client.get(
|
|
'/admin_tokens',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|
|
assert new_admin_token_name not in response.json()
|
|
|
|
|
|
def test_instance_state_updates(fastapi_client_simple):
|
|
test_client, _, admin_token = fastapi_client_simple
|
|
|
|
instance_state = get_instance_state()
|
|
|
|
forgejo_auth_spec = ForgejoAuthSpec(
|
|
type='forgejo',
|
|
url='http://localhost',
|
|
organization='orga-1',
|
|
team='team-1',
|
|
label_type='user',
|
|
)
|
|
|
|
forgejo_audit_collection_name = 'test_forgejo_audit_collection'
|
|
forgejo_audit_collection_request = CollectionRequest(
|
|
name=forgejo_audit_collection_name,
|
|
default_token='test_default_token',
|
|
curated=PurePosixPath(f'{curated}/{forgejo_audit_collection_name}'),
|
|
schema=test_schema_location,
|
|
incoming=PurePosixPath(f'{incoming}/{forgejo_audit_collection_name}'),
|
|
auth_sources=[forgejo_auth_spec],
|
|
)
|
|
|
|
# Post a collection with forgejo audit-backend without an instance_id
|
|
response = test_client.post(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=forgejo_audit_collection_request.model_dump(mode='json', by_alias=True),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
forgejo_instance = instance_state.auth_sources[forgejo_audit_collection_name][0]
|
|
assert forgejo_instance.instance_id is None
|
|
|
|
# Put a collection with updated forgejo audit-backend, containing an instance_id
|
|
forgejo_auth_spec.instance_id = 'instance-id'
|
|
response = test_client.put(
|
|
'/collections',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
json=forgejo_audit_collection_request.model_dump(mode='json', by_alias=True),
|
|
)
|
|
assert response.status_code == HTTP_201_CREATED
|
|
auth_sources = instance_state.auth_sources[forgejo_audit_collection_name]
|
|
assert len(auth_sources) == 1
|
|
new_forgejo_instance = auth_sources[0]
|
|
|
|
assert new_forgejo_instance is not forgejo_instance
|
|
assert new_forgejo_instance.instance_id == 'instance-id'
|
|
|
|
# Delete the collection again because we check for a known number of
|
|
# collections in other tests.
|
|
response = test_client.delete(
|
|
f'/collections/{forgejo_audit_collection_name}',
|
|
headers={'x-dumpthings-token': admin_token},
|
|
)
|
|
assert response.status_code == HTTP_200_OK
|