Skip to content

refactor: remove try catch blocks from tests #68

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arpad-csepi wants to merge 1 commit into
base: main
Choose a base branch
from
+84 −135
Open

refactor: remove try catch blocks from tests #68

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
219 changes: 84 additions & 135 deletions dir-sdk-python/agntcy/dir_sdk/tests/test_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,10 +71,7 @@ def test_publish(self) -> None:
record_refs=routing_v1.RecordRefs(refs=record_refs),
)

try:
self.client.publish(publish_request)
except Exception as e:
assert e is None
self.client.publish(publish_request)

def test_list(self) -> None:
records = self.gen_records(1, "list")
Expand Down Expand Up @@ -216,53 +213,43 @@ def test_unpublish(self) -> None:
_ = routing_v1.PublishRequest(record_refs=publish_record_refs)
unpublish_request = routing_v1.UnpublishRequest(record_refs=publish_record_refs)

try:
self.client.unpublish(unpublish_request)
except Exception as e:
assert e is None
self.client.unpublish(unpublish_request)

def test_delete(self) -> None:
records = self.gen_records(1, "delete")
record_refs = self.client.push(records=records)
try:
self.client.delete(record_refs)
except Exception as e:
assert e is None
self.client.delete(record_refs)

def test_push_referrer(self) -> None:
records = self.gen_records(2, "push_referrer")
record_refs = self.client.push(records=records)

try:
request = [
store_v1.PushReferrerRequest(
record_ref=record_refs[0],
type=sign_v1.Signature.DESCRIPTOR.full_name,
data={
"signature": "dGVzdC1zaWduYXR1cmU=", # base64 encoded "test-signature"
"annotations": {"payload": "test-payload-data"},
},
),
store_v1.PushReferrerRequest(
record_ref=record_refs[1],
type=sign_v1.Signature.DESCRIPTOR.full_name,
data={
"signature": "dGVzdC1zaWduYXR1cmU=", # base64 encoded "test-signature"
"annotations": {"payload": "test-payload-data"},
},
),
]

response = self.client.push_referrer(req=request)
request = [
store_v1.PushReferrerRequest(
record_ref=record_refs[0],
type=sign_v1.Signature.DESCRIPTOR.full_name,
data={
"signature": "dGVzdC1zaWduYXR1cmU=", # base64 encoded "test-signature"
"annotations": {"payload": "test-payload-data"},
},
),
store_v1.PushReferrerRequest(
record_ref=record_refs[1],
type=sign_v1.Signature.DESCRIPTOR.full_name,
data={
"signature": "dGVzdC1zaWduYXR1cmU=", # base64 encoded "test-signature"
"annotations": {"payload": "test-payload-data"},
},
),
]

assert response is not None
assert len(response) == 2
response = self.client.push_referrer(req=request)

for r in response:
assert isinstance(r, store_v1.PushReferrerResponse)
assert response is not None
assert len(response) == 2

except Exception as e:
assert e is None
for r in response:
assert isinstance(r, store_v1.PushReferrerResponse)

def test_pull_referrer(self) -> None:
records = self.gen_records(2, "pull_referrer")
Expand Down Expand Up @@ -293,42 +280,32 @@ def test_pull_referrer(self) -> None:
for r in response:
assert isinstance(r, store_v1.PushReferrerResponse)

try:
pull_request = [
store_v1.PullReferrerRequest(
record_ref=record_refs[0],
referrer_type=sign_v1.Signature.DESCRIPTOR.full_name,
),
store_v1.PullReferrerRequest(
record_ref=record_refs[1],
referrer_type=sign_v1.Signature.DESCRIPTOR.full_name,
),
]

pull_response = self.client.pull_referrer(req=pull_request)
pull_request = [
store_v1.PullReferrerRequest(
record_ref=record_refs[0],
referrer_type=sign_v1.Signature.DESCRIPTOR.full_name,
),
store_v1.PullReferrerRequest(
record_ref=record_refs[1],
referrer_type=sign_v1.Signature.DESCRIPTOR.full_name,
),
]

assert pull_response is not None
assert len(pull_response) == 2
pull_response = self.client.pull_referrer(req=pull_request)

for pull_r in pull_response:
assert isinstance(pull_r, store_v1.PullReferrerResponse)
except Exception as e:
assert "pull referrer not implemented" in str(
e,
) # Delete when the service implemented
assert pull_response is not None
assert len(pull_response) == 2

# self.assertIsNone(e) # Uncomment when the service implemented
for pull_r in pull_response:
assert isinstance(pull_r, store_v1.PullReferrerResponse)

def test_sign_and_verify(self) -> None:
records = self.gen_records(2, "sign_verify")
record_refs = self.client.push(records=records)

# Remove existing cosign keys if any
try:
pathlib.Path("cosign.key").unlink()
pathlib.Path("cosign.pub").unlink()
except FileNotFoundError:
pass # Clean state found
pathlib.Path("cosign.key").unlink(missing_ok=True)
pathlib.Path("cosign.pub").unlink(missing_ok=True)

# Prepare cosign key pair
key_password = "testing-key"
Expand Down Expand Up @@ -451,62 +428,45 @@ def test_sign_and_verify(self) -> None:
assert s_signer.oidc.issuer == r_signer.oidc.issuer
assert s_signer.oidc.subject == r_signer.oidc.subject

except Exception as e:
assert e is None, (
f"CID: {record_refs[0].cid} password: {key_provider.password!r} private_key: {key_provider.private_key}"
)
finally:
pathlib.Path("cosign.key").unlink()
pathlib.Path("cosign.pub").unlink()
pathlib.Path("cosign.key").unlink(missing_ok=True)
pathlib.Path("cosign.pub").unlink(missing_ok=True)

# Test invalid sign request
invalid_request = sign_v1.SignRequest(
record_ref=core_v1.RecordRef(cid="invalid-cid"),
provider=request_key_provider,
)
try:
with self.assertRaises(RuntimeError) as ctx:
self.client.sign(invalid_request)
except RuntimeError as e:
assert "Failed to sign the object" in str(e)
self.assertIn("Failed to sign the object", str(ctx.exception))

def test_sync(self) -> None:
try:
create_request = store_v1.CreateSyncRequest(
remote_directory_url=os.getenv(
"DIRECTORY_SERVER_PEER1_ADDRESS",
"0.0.0.0:8891",
),
)
create_response = self.client.create_sync(create_request)

try:
assert uuid.UUID(create_response.sync_id)
except ValueError as e:
msg = f"Not an UUID: {create_response.sync_id}"
raise ValueError(msg) from e
create_request = store_v1.CreateSyncRequest(
remote_directory_url=os.getenv(
"DIRECTORY_SERVER_PEER1_ADDRESS",
"0.0.0.0:8891",
),
)
create_response = self.client.create_sync(create_request)

list_request = store_v1.ListSyncsRequest()
list_response = self.client.list_syncs(list_request)
uuid.UUID(create_response.sync_id)

for sync_item in list_response:
try:
assert isinstance(sync_item, store_v1.ListSyncsItem)
assert uuid.UUID(sync_item.sync_id)
except ValueError as e:
msg = f"Not an UUID: {sync_item.sync_id}"
raise ValueError(msg) from e
list_request = store_v1.ListSyncsRequest()
list_response = self.client.list_syncs(list_request)

get_request = store_v1.GetSyncRequest(sync_id=create_response.sync_id)
get_response = self.client.get_sync(get_request)
for sync_item in list_response:
assert isinstance(sync_item, store_v1.ListSyncsItem)
uuid.UUID(sync_item.sync_id)

assert isinstance(get_response, store_v1.GetSyncResponse)
assert get_response.sync_id == create_response.sync_id
get_request = store_v1.GetSyncRequest(sync_id=create_response.sync_id)
get_response = self.client.get_sync(get_request)

delete_request = store_v1.DeleteSyncRequest(sync_id=create_response.sync_id)
self.client.delete_sync(delete_request)
assert isinstance(get_response, store_v1.GetSyncResponse)
assert get_response.sync_id == create_response.sync_id

except Exception as e:
assert e is None
delete_request = store_v1.DeleteSyncRequest(sync_id=create_response.sync_id)
self.client.delete_sync(delete_request)

def test_listen(self) -> None:
listen_request = events_v1.ListenRequest()
Expand Down Expand Up @@ -549,38 +509,27 @@ def test_publication(self) -> None:
records = self.gen_records(1, "publication")
record_refs = self.client.push(records=records)

try:
create_request = routing_v1.PublishRequest(
record_refs=routing_v1.RecordRefs(refs=record_refs),
)
create_request = routing_v1.PublishRequest(
record_refs=routing_v1.RecordRefs(refs=record_refs),
)

create_response = self.client.create_publication(create_request)
create_response = self.client.create_publication(create_request)

try:
assert isinstance(create_response, routing_v1.CreatePublicationResponse)
except ValueError as e:
msg = "Not a CreatePublicationResponse object."
raise ValueError(msg) from e

list_request = routing_v1.ListPublicationsRequest(limit=3)
list_response = self.client.list_publication(list_request)

for publication_item in list_response:
try:
assert isinstance(publication_item, routing_v1.ListPublicationsItem)
except ValueError as e:
msg = "Not a ListPublicationsItem object."
raise ValueError(msg) from e

get_request = routing_v1.GetPublicationRequest(
publication_id=create_response.publication_id
)
get_response = self.client.get_publication(get_request)
assert isinstance(create_response, routing_v1.CreatePublicationResponse)

list_request = routing_v1.ListPublicationsRequest(limit=3)
list_response = self.client.list_publication(list_request)

for publication_item in list_response:
assert isinstance(publication_item, routing_v1.ListPublicationsItem)

get_request = routing_v1.GetPublicationRequest(
publication_id=create_response.publication_id
)
get_response = self.client.get_publication(get_request)

assert isinstance(get_response, routing_v1.GetPublicationResponse)
assert get_response.publication_id == create_response.publication_id
except Exception as e:
assert e is None
assert isinstance(get_response, routing_v1.GetPublicationResponse)
assert get_response.publication_id == create_response.publication_id

def test_resolve(self) -> None:
# Push a record using built-in generator
Expand Down
Loading