Coverage for src/secchi/workflows/dashboard.py: 61%

1"""Dashboard project-resolution workflow.""" 

2 

3from dataclasses import dataclass 

4from pathlib import Path 

5 

6from secchi.config import find_config, load_project, load_projects 

7from secchi.diagnostics import DiagnosticLog 

8from secchi.errors import ConfigError, PackageNotFoundError 

9from secchi.models import PackageRef, Project 

10from secchi.services.resolver import parse_package_spec, resolve_package 

11 

12 

13@dataclass(frozen=True) 

14class DashboardRequest: 

15 project: Project 

16 config_path: Path 

17 workspace: list[Project] | None 

18 refresh: bool 

19 security_refresh: bool 

20 

21 

22def _workspace_project(config_path: Path) -> Project: 

23 projects = load_projects(config_path) 

24 refs: list[PackageRef] = [] 

25 for project in projects: 

26 refs.extend( 

27 PackageRef( 

28 ref.name, 

29 ref.registry, 

30 ref.favorite or project.favorite, 

31 project.name, 

32 ) 

33 for ref in project.packages 

34 ) 

35 return Project( 

36 name="Workspace", description="Configured Secchi workspace", packages=refs 

37 ) 

38 

39 

40async def run( 

41 *, 

42 package: str | None = None, 

43 registry: str | None = None, 

44 project_name: str | None = None, 

45 config: str | None = None, 

46 refresh: bool = False, 

47 security_refresh: bool = False, 

48 diagnostics: DiagnosticLog | None = None, 

49) -> DashboardRequest: 

50 config_path = find_config(config) 

51 workspace: list[Project] | None = None 

52 if package: 

53 ref = parse_package_spec(package, registry) 

54 configured = None 

55 if config_path and registry is None: 

56 for candidate in load_projects(config_path): 

57 if candidate.name.lower() == package.lower() or any( 

58 item.name.lower() == package.lower() for item in candidate.packages 

59 ): 

60 configured = candidate 

61 break 

62 if configured is not None: 

63 project = configured 

64 elif registry is None: 

65 refs = await resolve_package(package, diagnostics=diagnostics) 

66 if not refs: 

67 raise PackageNotFoundError( 

68 f"No exact package named '{package}' was found across registries." 

69 ) 

70 project = Project(name=ref.name, packages=refs) 

71 else: 

72 project = Project(name=ref.name, packages=[ref]) 

73 config_path = config_path or (Path.cwd() / "secchi.toml") 

74 else: 

75 if not config_path: 

76 raise ConfigError( 

77 "No config found. Use 'secchi dashboard PACKAGE' or create secchi.toml." 

78 ) 

79 if project_name: 

80 project = load_project(config_path, project_name) 

81 else: 

82 workspace = load_projects(config_path) 

83 project = _workspace_project(config_path) 

84 if not project.packages: 

85 raise ConfigError("The selected workspace has no packages.") 

86 return DashboardRequest( 

87 project, 

88 config_path, 

89 workspace if not package else None, 

90 refresh, 

91 security_refresh, 

92 )