Skip to content

Commit 7edd5f8

Browse files
BobroniumBobronium
committed
Add debug.py for ease of local development
1 parent 57ffcd6 commit 7edd5f8

File tree

2 files changed

+207
-0
lines changed

2 files changed

+207
-0
lines changed

‎packages/examples/cvat/exchange-oracle/debug.py‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
import datetime
2+
from pathlib import Path
3+
4+
import uvicorn
5+
6+
from src.chain.kvstore import register_in_kvstore
7+
from src.core.config import Config
8+
from src.services import cloud
9+
from src.services.cloud import BucketAccessInfo
10+
from src.utils.logging import get_function_logger
11+
12+
13+
def apply_local_development_patches():
14+
"""
15+
Applies local development patches to avoid manual source code modification.:
16+
- Overrides `EscrowUtils.get_escrow` to return local escrow data with mock values if the escrow
17+
address matches a local manifest.
18+
- Loads local manifest files from cloud storage into `LOCAL_MANIFEST_FILES`.
19+
- Disables address validation by overriding `validate_address`.
20+
- Replaces `validate_oracle_webhook_signature` with a lenient version that uses
21+
partial signature parsing.
22+
- Generates ECDSA keys if not already present for local JWT signing,
23+
and sets the public key in `Config.human_app_config`.
24+
"""
25+
from human_protocol_sdk.constants import ChainId
26+
from human_protocol_sdk.escrow import EscrowData, EscrowUtils
27+
28+
logger = get_function_logger(apply_local_development_patches.__name__)
29+
30+
minio_client = cloud.make_client(BucketAccessInfo.parse_obj(Config.storage_config))
31+
32+
def get_local_escrow(chain_id: int, escrow_address: str) -> EscrowData:
33+
possible_manifest_name = escrow_address.split(":")[0]
34+
local_manifests = minio_client.list_files(bucket="manifests")
35+
logger.info(f"Local manifests: {local_manifests}")
36+
if possible_manifest_name in local_manifests:
37+
return EscrowData(
38+
chain_id=ChainId(chain_id),
39+
id="test",
40+
address=escrow_address,
41+
amount_paid=10,
42+
balance=10,
43+
count=1,
44+
factory_address="",
45+
launcher="",
46+
status="Pending",
47+
token="HMT", # noqa: S106
48+
total_funded_amount=10,
49+
created_at=datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc),
50+
manifest_url=f"http://127.0.0.1:9010/manifests/{possible_manifest_name}",
51+
)
52+
return original_get_escrow(ChainId(chain_id), escrow_address)
53+
54+
original_get_escrow = EscrowUtils.get_escrow
55+
EscrowUtils.get_escrow = get_local_escrow
56+
57+
import src.schemas.webhook
58+
from src.core.types import OracleWebhookTypes
59+
60+
src.schemas.webhook.validate_address = lambda x: x
61+
62+
async def lenient_validate_oracle_webhook_signature(request, signature, webhook):
63+
from src.validators.signature import validate_oracle_webhook_signature
64+
65+
try:
66+
return OracleWebhookTypes(signature.split(":")[0])
67+
except (ValueError, TypeError):
68+
return validate_oracle_webhook_signature(request, signature, webhook)
69+
70+
import src.endpoints.webhook
71+
72+
src.endpoints.webhook.validate_oracle_webhook_signature = (
73+
lenient_validate_oracle_webhook_signature
74+
)
75+
import logging
76+
77+
logging.warning("Local development patches applied.")
78+
79+
from tests.api.test_exchange_api import generate_ecdsa_keys
80+
81+
# generating keys for local development
82+
repo_root = Path(__file__).parent
83+
human_app_private_key_file, human_app_public_key_file = (
84+
repo_root / "human_app_private_key.pem",
85+
repo_root / "human_app_public_key.pem",
86+
)
87+
if not (human_app_public_key_file.exists() and human_app_private_key_file.exists()):
88+
private_key, public_key = generate_ecdsa_keys()
89+
human_app_private_key_file.write_text(private_key)
90+
human_app_public_key_file.write_text(public_key)
91+
else:
92+
public_key = human_app_public_key_file.read_text()
93+
94+
Config.human_app_config.jwt_public_key = public_key
95+
96+
97+
if __name__ == "__main__":
98+
is_dev = Config.environment == "development"
99+
if is_dev:
100+
apply_local_development_patches()
101+
102+
Config.validate()
103+
register_in_kvstore()
104+
105+
uvicorn.run(
106+
app="src:app",
107+
host="0.0.0.0", # noqa: S104
108+
port=int(Config.port),
109+
workers=Config.workers_amount,
110+
)

‎packages/examples/cvat/recording-oracle/debug.py‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
import datetime
2+
3+
import uvicorn
4+
5+
from src.chain.kvstore import register_in_kvstore
6+
from src.core.config import Config
7+
from src.services import cloud
8+
from src.services.cloud import BucketAccessInfo
9+
from src.utils.logging import get_function_logger
10+
11+
12+
def apply_local_development_patches():
13+
"""
14+
Applies local development patches to bypass direct source code modifications:
15+
- Overrides `EscrowUtils.get_escrow` to retrieve local escrow data for specific addresses,
16+
using mock data if the address corresponds to a local manifest.
17+
- Updates local manifest files from cloud storage.
18+
- Overrides `validate_address` to disable address validation.
19+
- Replaces `validate_oracle_webhook_signature` with a lenient version for oracle signature
20+
validation in development.
21+
- Replaces `src.chain.escrow.store_results` to avoid attempting to store results on chain.
22+
- Replaces `src.validators.signature.validate_oracle_webhook_signature` to always return
23+
`OracleWebhookTypes.exchange_oracle`.
24+
"""
25+
logger = get_function_logger(apply_local_development_patches.__name__)
26+
27+
from human_protocol_sdk.constants import ChainId
28+
from human_protocol_sdk.escrow import EscrowData, EscrowUtils
29+
30+
minio_client = cloud.make_client(BucketAccessInfo.parse_obj(Config.storage_config))
31+
32+
def get_local_escrow(chain_id: int, escrow_address: str) -> EscrowData:
33+
possible_manifest_name = escrow_address.split(":")[0]
34+
local_manifests = minio_client.list_files(bucket="manifests")
35+
logger.info(f"Local manifests: {local_manifests}")
36+
if possible_manifest_name in local_manifests:
37+
return EscrowData(
38+
chain_id=ChainId(chain_id),
39+
id="test",
40+
address=escrow_address,
41+
amount_paid=10,
42+
balance=10,
43+
count=1,
44+
factory_address="",
45+
launcher="",
46+
status="Pending",
47+
token="HMT", # noqa: S106
48+
total_funded_amount=10,
49+
created_at=datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc),
50+
manifest_url=f"http://127.0.0.1:9010/manifests/{possible_manifest_name}",
51+
)
52+
return original_get_escrow(ChainId(chain_id), escrow_address)
53+
54+
original_get_escrow = EscrowUtils.get_escrow
55+
EscrowUtils.get_escrow = get_local_escrow
56+
57+
import src.schemas.webhook
58+
from src.core.types import OracleWebhookTypes
59+
60+
src.schemas.webhook.validate_address = lambda x: x
61+
62+
async def lenient_validate_oracle_webhook_signature(request, signature, webhook): # noqa: ARG001 (not relevant here)
63+
try:
64+
return OracleWebhookTypes(signature.split(":")[0])
65+
except (ValueError, TypeError):
66+
return OracleWebhookTypes.exchange_oracle
67+
68+
import src.endpoints.webhook
69+
70+
src.endpoints.webhook.validate_oracle_webhook_signature = (
71+
lenient_validate_oracle_webhook_signature
72+
)
73+
74+
import src.chain.escrow
75+
76+
def store_results(chain_id: int, escrow_address: str, url: str, hash: str) -> None: # noqa: ARG001 (not relevant here)
77+
logger.info(f"Would store results for escrow {escrow_address} on chain: {url}, {hash}")
78+
79+
src.chain.escrow.store_results = store_results
80+
81+
logger.warning("Local development patches applied.")
82+
83+
84+
if __name__ == "__main__":
85+
is_dev = Config.environment == "development"
86+
if is_dev:
87+
apply_local_development_patches()
88+
89+
Config.validate()
90+
register_in_kvstore()
91+
92+
uvicorn.run(
93+
app="src:app",
94+
host="0.0.0.0", # noqa: S104
95+
port=int(Config.port),
96+
workers=Config.workers_amount,
97+
)

0 commit comments

Comments
 (0)