Coverage for extra_codeowners/cli.py: 91%

1"""Extra CODEOWNERS command-line interface.""" 

2 

3from __future__ import annotations 

4 

5from pathlib import Path 

6from typing import Annotated 

7 

8import typer 

9import uvicorn 

10 

11from extra_codeowners.app import create_app 

12from extra_codeowners.codeowners import validate_pattern 

13from extra_codeowners.database import QueueStore 

14from extra_codeowners.models import OrganizationPolicy, RepositoryPolicy 

15from extra_codeowners.policy import compile_policy 

16from extra_codeowners.settings import Settings 

17 

18cli = typer.Typer( 

19 name="extra-codeowners", 

20 help="Run and validate the Extra CODEOWNERS GitHub App.", 

21 no_args_is_help=True, 

22) 

23 

24 

25@cli.command() 

26def serve( 

27 host: str | None = typer.Option(None, help="Bind address; overrides the environment."), 

28 port: int | None = typer.Option( 

29 None, min=1, max=65535, help="Port; overrides the environment." 

30 ), 

31) -> None: 

32 """Run the webhook API, durable worker, and reconciler.""" 

33 settings = Settings() 

34 updates: dict[str, object] = {} 

35 if host is not None: 35 ↛ 37line 35 didn't jump to line 37 because the condition on line 35 was always true

36 updates["host"] = host 

37 if port is not None: 37 ↛ 39line 37 didn't jump to line 39 because the condition on line 37 was always true

38 updates["port"] = port 

39 if updates: 39 ↛ 44line 39 didn't jump to line 44 because the condition on line 39 was always true

40 settings = settings.model_copy(update=updates) 

41 # Access logs are disabled because GitHub's one-use manifest conversion 

42 # code arrives in a callback query string. Structured application logs do 

43 # not record request URLs or secret values. 

44 uvicorn.run( 

45 create_app(settings), 

46 host=settings.host, 

47 port=settings.port, 

48 access_log=False, 

49 proxy_headers=True, 

50 ) 

51 

52 

53@cli.command("validate-policy") 

54def validate_policy( 

55 repository: Annotated[Path, typer.Option(exists=True, dir_okay=False, readable=True)], 

56 organization: Annotated[ 

57 Path | None, typer.Option(exists=True, dir_okay=False, readable=True) 

58 ] = None, 

59) -> None: 

60 """Compile repository and optional organization TOML policy files.""" 

61 repository_policy = RepositoryPolicy.from_toml(repository.read_text(encoding="utf-8")) 

62 if organization is not None: 

63 organization_policy = OrganizationPolicy.from_toml(organization.read_text(encoding="utf-8")) 

64 compile_policy(organization_policy, repository_policy) 

65 else: 

66 # Cross-file App enrollment requires --organization, but standalone 

67 # validation must still reject path syntax the runtime cannot compile. 

68 for delegation in repository_policy.delegations: 68 ↛ 71line 68 didn't jump to line 71 because the loop on line 68 didn't complete

69 for pattern in delegation.paths: 69 ↛ 68line 69 didn't jump to line 68 because the loop on line 69 didn't complete

70 validate_pattern(pattern) 

71 typer.echo("Policy files are valid.") 

72 

73 

74@cli.command("queue-status") 

75def queue_status() -> None: 

76 """Report pending jobs and any legacy terminal rows.""" 

77 settings = Settings() 

78 store = QueueStore(settings.database_url.get_secret_value()) 

79 try: 

80 store.initialize() 

81 typer.echo(f"pending={store.pending_count()} dead={store.dead_count()}") 

82 finally: 

83 store.close() 

84 

85 

86@cli.command("requeue-dead") 

87def requeue_dead( 

88 limit: Annotated[int, typer.Option(min=1, max=10_000)] = 100, 

89) -> None: 

90 """Recover a bounded batch of legacy/manual terminal rows.""" 

91 settings = Settings() 

92 store = QueueStore(settings.database_url.get_secret_value()) 

93 try: 

94 store.initialize() 

95 count = store.requeue_dead(limit) 

96 typer.echo(f"requeued={count}") 

97 finally: 

98 store.close() 

99 

100 

101def main() -> None: 

102 """Invoke the Typer command group.""" 

103 cli()