trentlarson/crowd-funder-from-jason
1
0
Fork 0
forked from jsnbuchanan/crowd-funder-for-time-pwa
Code Issues Pull Requests Projects Releases Wiki Activity

feat: add JSON output files and improve test flow

Browse Source 
- Add .generated directory for test artifacts
- Save key derivation data to key_derivation.json
- Save account initialization to account_init.json
- Save registration data to registration.json
- Save claim details to claim_details.json
- Save test environment to test-env.sh
- Save contacts data to contacts.json
- Add proper error handling for file operations
- Improve deeplink test flow with JSON-based data
- Add color output and better status messages
- Add ADB device detection and fallback to print mode

Technical Changes:
- Add file system operations with proper error handling
- Standardize JSON output format across Python/TypeScript
- Update test flow to use generated JSON files
- Add proper typing for registration response
- Improve error reporting and debug output

This improves the test workflow by saving all intermediate
data as JSON files that can be used by other test scripts.
The deeplink testing now uses this data instead of environment
variables for better reliability.
This commit is contained in:
Matthew Raymer
2025-03-08 13:01:15 +00:00
parent c57a7487e6
commit 9237f5a8d6
3 changed files with 401 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
test-scripts/new_flow.py
View File

@@ -54,6 +54,7 @@ from jwcrypto import jwk
import asyncio
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec
from pathlib import Path
# Constants
DEFAULT_ROOT_DERIVATION_PATH = "m/84737769'/0'/0'/0'" # Custom derivation path for TimeSafari
@@ -61,6 +62,16 @@ API_SERVER = "https://test-api.endorser.ch" # Endorser API endpoint
ENDORSER_DID = "did:ethr:0x0000694B58C2cC69658993A90D3840C560f2F51F" # Keep original case
ENDORSER_PRIVATE_KEY = "2b6472c026ec2aa2c4235c994a63868fc9212d18b58f6cbfe861b52e71330f5b"
# Create .generated directory if it doesn't exist
GENERATED_DIR = Path('.generated')
def initialize_directories():
"""Create necessary directories for storing generated files."""
try:
GENERATED_DIR.mkdir(exist_ok=True)
except Exception as e:
print(f"Error creating .generated directory: {e}")
raise
def derive_address(
mnemonic: str,
@@ -98,6 +109,21 @@ def derive_address(
print(f" Private Key: {private_hex[:8]}...")
print(f" Public Key: {public_hex[:8]}...")
# Save derivation data
derivation_data = {
"mnemonic": mnemonic,
"derivation_path": derivation_path,
"address": address,
"private_key": private_hex,
"public_key": public_hex
}
try:
GENERATED_DIR.mkdir(exist_ok=True)
with open(GENERATED_DIR / 'key_derivation.json', 'w') as f:
json.dump(derivation_data, f, indent=2)
except Exception as e:
print(f"Error saving derivation data: {e}")
return address, private_hex, public_hex, derivation_path
def new_identifier(
@@ -165,13 +191,16 @@ def initialize_account() -> Dict[str, Any]:
# Create DID identifier
identity = new_identifier(address, public_hex, private_hex, derivation_path)
# Format account data
# Save account data
account_data = {
"did": identity["did"],
"identity": identity,
"mnemonic": mnemonic,
"derivation_path": derivation_path
}
with open(GENERATED_DIR / 'account_init.json', 'w') as f:
json.dump(account_data, f, indent=2)
print("\nAccount initialized:")
print(json.dumps(account_data, indent=2))
print()
@@ -328,6 +357,15 @@ async def register(
response_data = response.json()
if response.status_code == 200 and response_data.get("success", {}).get("handleId"):
# Save registration data
registration_data = {
"active_did": active_did,
"jwt_token": jwt_token,
"response": response_data
}
with open(GENERATED_DIR / 'registration.json', 'w') as f:
json.dump(registration_data, f, indent=2)
return {
"success": True,
"handleId": response_data["success"]["handleId"],
@@ -393,15 +431,69 @@ async def fetch_claim(
headers={
'Authorization': f'Bearer {auth_token}',
'Content-Type': 'application/json'
}
},
timeout=30 # 30 second timeout
)
return response.json()
claim_data = response.json()
# Save claim data
try:
with open(GENERATED_DIR / 'claim_details.json', 'w', encoding='utf-8') as f:
json.dump({
'claim_id': claim_id,
'active_did': active_did,
'response': claim_data
}, f, indent=2)
except Exception as e:
print(f"Error saving claim data: {e}")
return claim_data
except requests.RequestException as e:
print("\nError fetching claim:")
print(f"Error: {str(e)}")
# Save error state
try:
with open(
GENERATED_DIR / 'claim_details.json',
'w', encoding='utf-8') as f:
json.dump({
'claim_id': claim_id,
'active_did': active_did,
'error': str(e)
}, f, indent=2)
except Exception as write_err:
print(f"Error saving claim error: {write_err}")
raise
async def generate_test_env():
"""Generate test environment data for deeplink testing"""
test_env_data = {
'CONTACT1_DID': active_did,
'CONTACT1_KEY': private_key_hex,
'CONTACT2_DID': recipient_did,
'CONTACT2_KEY': recipient_key,
'ISSUER_DID': issuer_did,
'ISSUER_KEY': issuer_key,
'TEST_JWT': jwt_token,
'API_SERVER': API_SERVER
}
# Write test environment variables
with open(GENERATED_DIR / 'test-env.sh', 'w', encoding='utf-8') as f:
for key, value in test_env_data.items():
f.write(f'export {key}="{value}"\n')
# Write test contacts data
contacts_data = {
'contacts': [
{'did': active_did, 'name': 'Test Contact 1'},
{'did': recipient_did, 'name': 'Test Contact 2'}
]
}
with open(GENERATED_DIR / 'contacts.json', 'w', encoding='utf-8') as f:
json.dump(contacts_data, f, indent=2)
# Main execution
async def main():
"""
@@ -442,28 +534,38 @@ async def main():
Registration result: {"success": true}
```
"""
# Step 1: Create a new DID
identity = initialize_account()
active_did = identity["did"]
private_key_hex = identity["keys"][0]["privateKeyHex"]
# Initialize directories first
initialize_directories()
# Step 2: Register the DID
result = await register(active_did, private_key_hex)
print("Registration result:", result)
try:
# Step 1: Create a new DID
identity = initialize_account()
active_did = identity["did"]
private_key_hex = identity["keys"][0]["privateKeyHex"]
# Step 3: If registration successful, fetch claim details
if result.get("success") and result.get("claimId"):
print("\nFetching claim details...")
try:
claim_details = await fetch_claim(
result["claimId"],
active_did,
private_key_hex
)
print("\nClaim Details:")
print(json.dumps(claim_details, indent=2))
except Exception as e:
print(f"Error fetching claim: {str(e)}")
# Step 2: Register the DID
result = await register(active_did, private_key_hex)
print("Registration result:", result)
# Step 3: If registration successful, fetch claim details
if result.get("success") and result.get("claimId"):
print("\nFetching claim details...")
try:
claim_details = await fetch_claim(
result["claimId"],
active_did,
private_key_hex
)
print("\nClaim Details:")
print(json.dumps(claim_details, indent=2))
except Exception as e:
print(f"Error fetching claim: {str(e)}")
# Step 4: Generate test environment data
await generate_test_env()
except Exception as e:
print(f"Error: {str(e)}")
if __name__ == "__main__":
asyncio.run(main())