claudi
389d72a136
- Implement tests for `EbayOAuthClient` to verify authorization URL generation with configured scopes and token reuse logic. - Add tests for `WebhookChallengeHandler` to ensure correct SHA256 response generation and for `WebhookSignatureParser` to validate extraction of known fields from signature strings.
134 lines
4.2 KiB
Python
134 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
GENERATED_ROOT = ROOT / "ebay_client" / "generated"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ApiSpec:
|
|
name: str
|
|
spec_path: Path
|
|
output_path: Path
|
|
project_name: str
|
|
package_name: str
|
|
|
|
|
|
API_SPECS = {
|
|
"notification": ApiSpec(
|
|
name="notification",
|
|
spec_path=ROOT / "commerce_notification_v1_oas3.yaml",
|
|
output_path=GENERATED_ROOT / "notification",
|
|
project_name="ebay-notification-generated",
|
|
package_name="notification_generated",
|
|
),
|
|
"inventory": ApiSpec(
|
|
name="inventory",
|
|
spec_path=ROOT / "sell_inventory_v1_oas3.yaml",
|
|
output_path=GENERATED_ROOT / "inventory",
|
|
project_name="ebay-inventory-generated",
|
|
package_name="inventory_generated",
|
|
),
|
|
"fulfillment": ApiSpec(
|
|
name="fulfillment",
|
|
spec_path=ROOT / "sell_fulfillment_v1_oas3.yaml",
|
|
output_path=GENERATED_ROOT / "fulfillment",
|
|
project_name="ebay-fulfillment-generated",
|
|
package_name="fulfillment_generated",
|
|
),
|
|
"account": ApiSpec(
|
|
name="account",
|
|
spec_path=ROOT / "sell_account_v1_oas3.yaml",
|
|
output_path=GENERATED_ROOT / "account",
|
|
project_name="ebay-account-generated",
|
|
package_name="account_generated",
|
|
),
|
|
"feed": ApiSpec(
|
|
name="feed",
|
|
spec_path=ROOT / "sell_feed_v1_oas3.yaml",
|
|
output_path=GENERATED_ROOT / "feed",
|
|
project_name="ebay-feed-generated",
|
|
package_name="feed_generated",
|
|
),
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Generate isolated low-level clients from the eBay OpenAPI contracts.")
|
|
parser.add_argument("--api", choices=sorted(API_SPECS), help="Generate only one API package.")
|
|
parser.add_argument("--fail-on-warning", action="store_true", help="Fail if the generator emits warnings.")
|
|
return parser.parse_args()
|
|
|
|
|
|
def build_config(spec: ApiSpec) -> dict[str, object]:
|
|
return {
|
|
"project_name_override": spec.project_name,
|
|
"package_name_override": spec.package_name,
|
|
"field_prefix": "field_",
|
|
"generate_all_tags": False,
|
|
"http_timeout": 30,
|
|
"literal_enums": False,
|
|
"post_hooks": [],
|
|
"use_path_prefixes_for_title_model_names": False,
|
|
}
|
|
|
|
|
|
def run_generation(spec: ApiSpec, *, fail_on_warning: bool) -> None:
|
|
if not spec.spec_path.exists():
|
|
raise FileNotFoundError(f"OpenAPI spec not found: {spec.spec_path}")
|
|
|
|
with tempfile.TemporaryDirectory(prefix=f"ebay_{spec.name}_") as temp_dir_name:
|
|
temp_dir = Path(temp_dir_name)
|
|
config_path = temp_dir / "config.yaml"
|
|
temp_output = temp_dir / "output"
|
|
config_path.write_text(yaml.safe_dump(build_config(spec), sort_keys=True), encoding="utf-8")
|
|
|
|
command = [
|
|
sys.executable,
|
|
"-m",
|
|
"openapi_python_client",
|
|
"generate",
|
|
"--path",
|
|
str(spec.spec_path),
|
|
"--meta",
|
|
"none",
|
|
"--overwrite",
|
|
"--config",
|
|
str(config_path),
|
|
"--output-path",
|
|
str(temp_output),
|
|
]
|
|
if fail_on_warning:
|
|
command.append("--fail-on-warning")
|
|
|
|
subprocess.run(command, check=True, cwd=str(ROOT))
|
|
|
|
generated_package = temp_output / spec.package_name
|
|
if not generated_package.exists():
|
|
generated_package = temp_output
|
|
if spec.output_path.exists():
|
|
shutil.rmtree(spec.output_path)
|
|
shutil.copytree(generated_package, spec.output_path)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
specs = [API_SPECS[args.api]] if args.api else [API_SPECS[name] for name in sorted(API_SPECS)]
|
|
for spec in specs:
|
|
print(f"Generating {spec.name} from {spec.spec_path.name} -> {spec.output_path}")
|
|
run_generation(spec, fail_on_warning=args.fail_on_warning)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|