refactor: format refactored tests with black to fix CI

Ran `poetry run black --check .` and `black src/tests/test_nostr_index_size.py src/tests/test_nostr_sdk_workflow.py`
to fix code formatting issues in the test audit run, which caused GitHub
Actions CI checks to fail.

Co-authored-by: PR0M3TH3AN <[email protected]>
This commit is contained in:
google-labs-jules[bot]andPR0M3TH3AN committed 2026-03-01 03:09:02 +00:00
1 parent e1400f32a5
commit 6e916fdb69
2 files changed
+16 -7

No files matched your search

+16 -5
View File
@@ -84,26 +84,37 @@ def test_nostr_index_size_limits(pytestconfig: pytest.Config):
encrypted = vault.get_encrypted_index()
payload_size = len(encrypted) if encrypted else 0
asyncio.run(client.publish_snapshot(encrypted or b""))
async def fetch_with_retry(client, expected_data, timeout, interval=1.0):
async def fetch_with_retry(
client, expected_data, timeout, interval=1.0
):
start_time = time.time()
while time.time() - start_time < timeout:
result = await client.fetch_latest_snapshot()
retrieved = gzip.decompress(b"".join(result[1])) if result else None
retrieved = (
gzip.decompress(b"".join(result[1])) if result else None
)
if retrieved == expected_data:
return True
await asyncio.sleep(interval)
return False
retrieved_ok = asyncio.run(fetch_with_retry(client, encrypted, delay))
retrieved_ok = asyncio.run(
fetch_with_retry(client, encrypted, delay)
)
if not retrieved_ok:
print(f"Initial retrieve failed: {client.last_error}")
retrieved_ok = asyncio.run(fetch_with_retry(client, encrypted, delay))
retrieved_ok = asyncio.run(
fetch_with_retry(client, encrypted, delay)
)
if not retrieved_ok:
print("Trying alternate relay")
client.update_relays(["wss://relay.damus.io"])
retrieved_ok = asyncio.run(fetch_with_retry(client, encrypted, delay))
retrieved_ok = asyncio.run(
fetch_with_retry(client, encrypted, delay)
)
results.append((entry_count, payload_size, True, retrieved_ok))
if max_entries is not None:
if entry_count >= max_entries:
-2
View File
@@ -40,12 +40,10 @@ def test_nostr_sdk_send_receive(tmp_path):
thread = threading.Thread(target=run_relay, args=(relay,), daemon=True)
thread.start()
seed = "abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about"
fingerprint = "test"
km = KeyManager(seed, fingerprint)
# Wait for the relay to be ready
ws = None
for _ in range(40):