from pathlib import Path import copy import sys import unittest ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import console_runner class QueryOverrideTests(unittest.TestCase): def source(self): return { 'mode': 'search', 'pages': 20, 'per_page': 10, 'max_targets': 0, 'workers': 2, 'query_overrides': { 'openai': {'pages': 2, 'per_page': 10, 'max_targets': 20}, }, } def test_exact_query_override_is_applied_without_mutating_source(self): source = self.source() original = copy.deepcopy(source) args = console_runner.build_args_from_source_config( 'dockerhub', source, {}, 'openai', ) self.assertEqual((args.pages, args.per_page, args.max_targets), (2, 10, 20)) self.assertEqual(source, original) def test_ordinary_query_retains_source_defaults(self): args = console_runner.build_args_from_source_config( 'dockerhub', self.source(), {}, 'chatgpt', ) self.assertEqual((args.pages, args.per_page, args.max_targets), (20, 10, 0)) def test_invalid_query_override_fails_closed(self): source = self.source() source['query_overrides']['openai'] = {'workers': 99} with self.assertRaisesRegex(ValueError, 'unsupported keys: workers'): console_runner.build_args_from_source_config( 'dockerhub', source, {}, 'openai', ) source['query_overrides']['openai'] = 'unbounded' with self.assertRaisesRegex(ValueError, 'must be a mapping'): console_runner.build_args_from_source_config( 'dockerhub', source, {}, 'openai', ) class CanonicalOpenAICoverageTests(unittest.TestCase): def test_retired_exact_query_and_overrides_are_absent(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) for source_name in ('github', 'gitlab', 'dockerhub'): source = config['sources'][source_name] self.assertNotIn('openai', source['queries']) self.assertNotIn('openai', source['query_overrides']) self.assertEqual(config['sources']['gitlab']['updated_target_rescan_max_per_cycle'], 1) self.assertTrue(config['sources']['dockerhub']['require_digest']) self.assertFalse(config['sources']['dockerhub'].get('updated_target_rescan_enabled', False)) def test_ecosystem_queries_are_source_specific_unique_and_bounded(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) expected = { 'github': { 'OPENAI_API_KEY': {'pages': 1, 'per_page': 25, 'max_targets': 5}, 'api.openai.com': {'pages': 1, 'per_page': 25, 'max_targets': 5}, 'openai-agents': {'pages': 1, 'per_page': 50, 'max_targets': 5}, }, 'gitlab': { 'openai-api': {'pages': 1, 'per_page': 50, 'max_targets': 5}, 'openai-agents': {'pages': 1, 'per_page': 50, 'max_targets': 5}, 'librechat': {'pages': 1, 'per_page': 25, 'max_targets': 5}, }, 'dockerhub': { 'librechat': {'max_targets': 10}, 'lobechat': {'max_targets': 10}, 'openai-proxy': {'max_targets': 10}, }, } for source_name, queries in expected.items(): source = config['sources'][source_name] self.assertEqual(len(source['queries']), len(set(source['queries']))) for query, bounds in queries.items(): self.assertEqual(source['queries'].count(query), 1) self.assertEqual(source['query_overrides'][query], bounds) args = console_runner.build_args_from_source_config( source_name, source, config['global'], query, ) self.assertEqual( (args.pages, args.per_page, args.max_targets), ( bounds.get('pages', source['pages']), bounds.get('per_page', source['per_page']), bounds['max_targets'], ), ) self.assertNotIn('OPENAI_API_KEY', config['sources']['gitlab']['queries']) self.assertNotIn('api.openai.com', config['sources']['dockerhub']['queries']) for source_name in expected: source_queries = config['sources'][source_name]['queries'] for retained_query in ('llm', 'ai'): self.assertEqual(source_queries.count(retained_query), 1) for retired_query in ('chatgpt', 'gpt', 'model'): self.assertNotIn(retired_query, source_queries) self.assertEqual(config['sources']['gitlab']['updated_target_rescan_max_per_cycle'], 1) self.assertEqual(config['sources']['gitlab']['updated_target_rescan_cooldown_hours'], 24) self.assertTrue(config['sources']['dockerhub']['require_digest']) def test_zero_yield_keyword_cohort_is_pruned_consistently(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) common = { 'autonomous', 'benchmarks', 'claw', 'code-assistant', 'codegen', 'dspy', 'embedding', 'embeddings', 'eval', 'evals', 'gateway', 'grok', 'haystack', 'inference', 'inference-api', 'knowledge', 'llamaindex', 'model', 'model-router', 'models', 'ollama', 'orchestration', 'prompts', 'replicate', 'rerank', 'reranker', 'retrieval', 'router', 'tokenizer', 'tool-use', 'vector', 'vllm', } shared = {'chatgpt', 'gpt', 'moonshot'} expected_sizes = { 'github': 62, 'gitlab': 51, 'dockerhub': 61, 'npm': 19, 'pypi': 41, 'package_git': 36, 'postman': 32, } for source_name in ('github', 'gitlab', 'dockerhub', 'npm', 'pypi', 'package_git'): source = config['sources'][source_name] queries = source['queries'] self.assertTrue((common | shared).isdisjoint(queries)) self.assertEqual(len(queries), expected_sizes[source_name]) self.assertEqual(len(queries), len(set(queries))) postman = config['sources']['postman']['queries'] self.assertTrue((shared | { 'openai', 'dashscope-intl.aliyuncs.com', 'generativelanguage.googleapis.com', }).isdisjoint(postman)) self.assertEqual(len(postman), expected_sizes['postman']) self.assertEqual(len(postman), len(set(postman))) self.assertEqual(config['sources']['github_archive']['queries'], ['gharchive']) self.assertEqual(config['sources']['github_archive_files']['queries'], ['gharchive-files']) self.assertEqual(config['sources']['github_gists']['queries'], ['gists']) for source_name in expected_sizes: source = config['sources'][source_name] overrides = set(source.get('query_overrides') or {}) self.assertTrue(overrides.issubset(source['queries'])) for source_name in ('github', 'gitlab', 'dockerhub'): queries = config['sources'][source_name]['queries'] for productive_query in ('llm', 'ai', 'agents', 'chat', 'prompt', 'rag', 'langchain', 'groq'): self.assertIn(productive_query, queries) def test_experimental_app_queries_are_source_specific_and_bounded(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) queries = ['hermes', 'harnes', 'flow', 'helpdesk', 'paperless'] expected = { 'gitlab': ({'pages': 1, 'per_page': 50, 'max_targets': 5}, (1, 50, 5)), 'dockerhub': ({'max_targets': 10}, (30, 100, 10)), } for source_name, (override, effective) in expected.items(): source = config['sources'][source_name] start = source['queries'].index(queries[0]) self.assertEqual(source['queries'][start:start + len(queries)], queries) for query in queries: self.assertEqual(source['queries'].count(query), 1) self.assertEqual(source['query_overrides'][query], override) args = console_runner.build_args_from_source_config( source_name, source, config['global'], query, ) self.assertEqual( (args.pages, args.per_page, args.max_targets), effective, ) self.assertEqual(config['sources']['huggingface']['queries'], ['spaces']) def test_dockerhub_discovery_policy_is_exact_unique_and_uniform(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) source = config['sources']['dockerhub'] existing_queries = [ 'llm', 'ai', 'agents', 'assistant', 'bot', 'chatbot', 'rag', 'semantic', 'prompt', 'completion', 'completions', 'mcp', 'openrouter', 'groq', 'xai', 'langchain', 'litellm', 'gemini', 'qwen', 'kimi', 'dashscope', 'aistudio', 'studio', 'librechat', 'lobechat', 'openai-proxy', 'open', 'chat', 'chats', 'conversation', 'conversational', 'dialogue', 'copilot', 'coding', 'ai-assistant', 'ai-agent', 'multi-agent', 'workflow', 'workflows', 'llmops', 'benchmark', 'tokens', 'memory', 'function-calling', 'hermes', 'harnes', 'flow', 'helpdesk', 'paperless', ] confirmed_tail = [ 'open-webui', 'ragflow', 'dify', 'flowise', 'crewai', 'n8n', 'langflow', 'autogen', 'browser-use', 'openhands', 'anythingllm', 'agent-zero', ] expected_overrides = { query: {'max_targets': 10} for query in ( 'librechat', 'lobechat', 'openai-proxy', 'hermes', 'harnes', 'flow', 'helpdesk', 'paperless', ) } self.assertEqual(source['queries'], existing_queries + confirmed_tail) self.assertEqual(len(source['queries']), 61) self.assertEqual(source['queries'][-len(confirmed_tail):], confirmed_tail) self.assertEqual(len(source['queries']), len(set(source['queries']))) self.assertEqual((source['pages'], source['per_page']), (30, 100)) self.assertEqual(source['query_overrides'], expected_overrides) for query in source['queries']: args = console_runner.build_args_from_source_config( 'dockerhub', source, config['global'], query, ) self.assertEqual((args.pages, args.per_page), (30, 100), query) self.assertEqual( args.max_targets, 10 if query in expected_overrides else 0, query, ) def test_dockerhub_refresh_and_resource_retry_policy_is_pinned(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) source = config['sources']['dockerhub'] expected_source_policy = { 'require_digest': True, 'auth_rotation': 'per_scan', 'retry_with_next_auth_on_rate_limit': True, 'rate_limit_cooldown': 1800, 'workers': 2, 'trufflehog_job_memory_limit_bytes': 6442450944, 'timeout': 600, 'trufflehog_concurrency': 4, 'fetch_workers': 5, 'fetch_timeout': 15, 'tag_fetch_workers': 4, 'tag_retry_count': 0, 'tag_retry_delay': 10, 'tag_resolve_limit': 100, 'docker_platform_filter_enabled': True, 'docker_platform_os': 'linux', 'docker_platform_arch': 'amd64', 'docker_platform_candidate_tags': 20, 'docker_images_per_repository': 3, 'docker_content_scan_mode': 'canary', 'docker_layer_canary_basis_points': 10000, 'docker_layer_config_max_bytes': 1048576, 'docker_layer_max_bytes': 268435456, 'docker_layer_image_max_bytes': 1073741824, 'docker_layer_max_layers': 8, 'docker_layer_archive_max_size_bytes': 268435456, 'docker_layer_archive_max_depth': 4, 'docker_layer_archive_timeout_sec': 30, 'docker_layer_blob_timeout_sec': 600, 'docker_layer_filesystem_concurrency': 2, 'docker_layer_blob_max_attempts': 3, 'docker_layer_blob_lease_sec': 1800, 'docker_layer_min_free_bytes': 21474836480, 'docker_layer_checkpoint_delay_sec': 60, } expected_retry_policy = { 'target_retry_max_attempts': 3, 'target_retry_base_delay_sec': 3600, 'target_retry_max_delay_sec': 86400, 'target_timeout_retry_delay_sec': 21600, } self.assertTrue(source['refresh_registry']) self.assertEqual(source['docker_repository_refresh_interval_sec'], 86400) self.assertEqual(source['docker_repository_refresh_max_per_cycle'], 0) self.assertEqual( {key: source[key] for key in expected_source_policy}, expected_source_policy, ) self.assertEqual( {key: config['global'][key] for key in expected_retry_policy}, expected_retry_policy, ) if __name__ == '__main__': unittest.main()