import json import os from pathlib import Path import sys import tempfile from types import SimpleNamespace import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[1] APP_DIR = ROOT / 'app' sys.path.insert(0, str(APP_DIR)) import console_runner import scanner class CandidateQueueCapacityTests(unittest.TestCase): @classmethod def setUpClass(cls): scanner.initialize_scanner_runtime(preflight_complete=True, register_cleanup=False) @staticmethod def write_private(path, text): Path(path).write_text(text, encoding='utf-8') scanner.harden_private_file(path) def candidate_limits(self, items=2): return ( mock.patch.object(scanner.scan_config, 'keycheck_candidate_file_max_items', items), mock.patch.object(scanner.scan_config, 'keycheck_candidate_file_max_bytes', 4096), mock.patch.object(scanner.scan_config, 'keycheck_candidate_line_max_bytes', 1024), ) def test_saturated_unconsumed_queue_raises_without_partial_append(self): with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) path = os.path.join(temp_dir, 'gem.txt') self.write_private(path, 'first\torigin-a\nsecond\torigin-b\n') original = Path(path).read_bytes() patches = self.candidate_limits() with patches[0], patches[1], patches[2]: with self.assertRaises(scanner.CandidateQueueCapacityError): scanner.append_unique_lines_locked( path, ['third\torigin-c', 'fourth\torigin-d'], ) self.assertEqual(Path(path).read_bytes(), original) def test_checked_rows_compact_then_complete_batch_appends(self): cases = ( ('gem.txt', 'geminiChecked.txt'), ('azureOpenAI.txt', 'azureChecked.txt'), ('azureFoundry.txt', 'azureChecked.txt'), ) for candidate_name, checked_name in cases: with self.subTest(candidate=candidate_name), tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) path = os.path.join(temp_dir, candidate_name) checked = os.path.join(temp_dir, checked_name) self.write_private(path, 'first\torigin-a\nsecond\torigin-b\n') self.write_private(checked, 'first\tINVALID\t2026-07-19\nsecond\tVALID\t2026-07-19\n') patches = self.candidate_limits() with patches[0], patches[1], patches[2]: inserted = scanner.append_unique_lines_locked( path, ['third\torigin-c', 'fourth\torigin-d'], ) self.assertEqual(inserted, 2) self.assertEqual( Path(path).read_text(encoding='utf-8').splitlines(), ['third\torigin-c', 'fourth\torigin-d'], ) def test_duplicate_identity_only_is_successful_noop_at_capacity(self): with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) path = os.path.join(temp_dir, 'gem.txt') self.write_private(path, 'first\torigin-a\nsecond\torigin-b\n') original = Path(path).read_bytes() patches = self.candidate_limits() with patches[0], patches[1], patches[2]: inserted = scanner.append_unique_lines_locked(path, ['first\tnew-origin']) self.assertEqual(inserted, 0) self.assertEqual(Path(path).read_bytes(), original) def test_structured_publication_reports_capacity_failure(self): result = { 'structured_keycheck_pending': True, 'postman': {'cache_path': 'fixture.json'}, 'bytes': 10, 'postman_max_artifact_size_mb': 1, } error = scanner.CandidateQueueCapacityError('candidate queue full') with mock.patch.object( console_runner, 'validate_postman_cache_artifact', return_value=('fixture.json', 10), ), mock.patch.object( console_runner, 'write_structured_keycheck_candidates', side_effect=error, ), mock.patch.object(console_runner, 'save_scan_result') as save: ok, message = console_runner.publish_scan_payload(result) self.assertFalse(ok) self.assertIn('candidate queue full', message) save.assert_not_called() def test_finding_candidate_capacity_makes_result_save_fail(self): result = { 'scan_event_id': 'capacity-event', 'target': 'fixture-target', 'scan_type': 'github', 'timestamp': '2026-07-19T00:00:00+00:00', 'findings': [{'DetectorName': 'AzureFoundry', 'Raw': 'fixture'}], 'errors': [], } error = scanner.CandidateQueueCapacityError('foundry queue full') with tempfile.TemporaryDirectory() as temp_dir: scanner.ensure_private_directory(temp_dir, reject_reparse=True) with mock.patch.object( scanner.scan_config, 'results_dir', temp_dir, ), mock.patch.object( scanner, 'write_foundry_keycheck_candidates_from_findings', side_effect=error, ), mock.patch.object( scanner, 'append_rotating_jsonl_once', return_value=True, ): self.assertFalse(scanner.save_scan_result(result)) class BoundedKnownTargetTests(unittest.TestCase): class Response: status_code = 200 headers = {} def __init__(self, items): self.items = items def raise_for_status(self): return None def json(self): return {'items': self.items} def test_postgres_early_stop_lookup_receives_only_current_page(self): class DB: postgres_required = True conn = SimpleNamespace(is_postgres=True) def __init__(self): self.calls = [] def known_target_normalizations(self, source, platform): raise AssertionError('broad history lookup must not be called') def known_target_normalizations_for(self, source, platform, targets): self.calls.append((source, platform, list(targets))) return {console_runner.normalize_target(target, platform) for target in targets} db = DB() args = SimpleNamespace( platform='github', database_url='postgresql://fixture/db', stop_on_seen_pages=True, seen_page_threshold=1, min_pages_before_stop=1, ) page = [ {'clone_url': 'https://github.com/acme/one.git', 'full_name': 'acme/one'}, {'clone_url': 'https://github.com/acme/two.git', 'full_name': 'acme/two'}, ] options = console_runner.fetch_stop_options(args, db, 'source-name') with mock.patch.object( scanner, 'api_request', return_value=self.Response(page), ) as request: scanner.fetch_github_repo_items('fixture', 3, per_page=2, **options) self.assertEqual(request.call_count, 1) self.assertEqual(len(db.calls), 1) self.assertEqual(db.calls[0][:2], ('source-name', 'github')) self.assertEqual( db.calls[0][2], ['https://github.com/acme/one.git', 'https://github.com/acme/two.git'], ) def test_updated_rescan_preserves_each_source_timestamp_and_disables_identity_stop(self): fixtures = [ ('github', 'pushed_at', '2026-08-25T10:00:00Z'), ('gitlab', 'last_activity_at', '2026-08-25T11:00:00Z'), ('huggingface', 'updated_at', '2026-08-25T12:00:00Z'), ] for platform, field, timestamp in fixtures: with self.subTest(platform=platform): args = SimpleNamespace( platform=platform, max_repo_age_days=0, updated_target_rescan_enabled=True, stop_on_seen_pages=True, ) item = {'url': f'https://example.invalid/{platform}', field: timestamp} self.assertEqual( console_runner.filter_repo_items_by_age([item], args), [{ 'target': item['url'], 'remote_modified_at': timestamp.replace('Z', '+00:00'), }], ) self.assertEqual(console_runner.fetch_stop_options(args), {}) docker_args = SimpleNamespace( platform='docker', max_repo_age_days=0, updated_target_rescan_enabled=True, ) self.assertEqual( console_runner.filter_repo_items_by_age( [{'url': 'repo/image', 'updated_at': '2026-08-25T13:00:00Z'}], docker_args, ), ['repo/image'], ) def test_github_updated_rescan_forces_metadata_discovery(self): args = SimpleNamespace( platform='github', mode='search', token=None, query='fixture', pages=1, per_page=10, sort_by='updated', sort_order='desc', created_filter='any', max_repo_age_days=0, updated_target_rescan_enabled=True, raise_rate_limit=False, stop_on_seen_pages=True, ) item = { 'url': 'https://github.com/acme/repo.git', 'pushed_at': '2026-08-25T13:00:00Z', } with mock.patch.object(console_runner, 'fetch_github_repo_items', return_value=[item]) as metadata, \ mock.patch.object(console_runner, 'fetch_github_repos') as urls: self.assertEqual( console_runner.fetch_targets(args), [{ 'target': item['url'], 'remote_modified_at': item['pushed_at'].replace('Z', '+00:00'), }], ) metadata.assert_called_once() urls.assert_not_called() def test_updated_rescan_keeps_new_targets_with_missing_or_invalid_age(self): args = SimpleNamespace( platform='gitlab', max_repo_age_days=1, repo_age_field='last_activity_at', updated_target_rescan_enabled=True, ) items = [ {'url': 'https://gitlab.com/acme/missing.git'}, { 'url': 'https://gitlab.com/acme/invalid.git', 'last_activity_at': 'not-a-time', }, ] self.assertEqual( console_runner.filter_repo_items_by_age(items, args), [ {'target': items[0]['url'], 'remote_modified_at': None}, {'target': items[1]['url'], 'remote_modified_at': None}, ], ) def test_huggingface_updated_rescan_uses_newest_modified_cursor(self): class Response: status_code = 200 def __init__(self, items, next_url=''): self.items = items self.links = {'next': {'url': next_url}} if next_url else {} def raise_for_status(self): return None def json(self): return self.items responses = [ Response([{'id': 'acme/one', 'lastModified': '2026-08-25T12:00:00Z'}], 'https://huggingface.co/api/spaces?cursor=next'), Response([{'id': 'acme/two', 'lastModified': 'not-a-time'}]), ] with mock.patch.object(scanner, 'api_request', side_effect=responses) as request: spaces = scanner.fetch_huggingface_spaces( pages=3, request_timeout=5, return_metadata=True, request_attempts=3, retry_delay=5, ) self.assertEqual([item['url'] for item in spaces], ['acme/one', 'acme/two']) self.assertEqual(spaces[0]['updated_at'], '2026-08-25T12:00:00Z') self.assertEqual(request.call_count, 2) self.assertEqual(request.call_args_list[0].kwargs['params'], { 'sort': 'lastModified', 'direction': '-1', 'limit': 100, }) self.assertIsNone(request.call_args_list[1].kwargs['params']) self.assertEqual(request.call_args_list[0].kwargs['max_retries'], 3) self.assertEqual(request.call_args_list[0].kwargs['retry_delay'], 5) def test_huggingface_discovery_retries_a_transient_timeout(self): class Response: status_code = 200 links = {} def raise_for_status(self): return None def json(self): return [{'id': 'acme/space'}] timeout = scanner.requests.exceptions.ReadTimeout('fixture timeout') with mock.patch.object(scanner.scan_config, 'api_proxy_enabled', False), \ mock.patch.object( scanner.requests, 'request', side_effect=[timeout, Response()], ) as request: spaces = scanner.fetch_huggingface_spaces( pages=1, request_timeout=5, return_metadata=True, request_attempts=3, retry_delay=0, ) self.assertEqual([item['url'] for item in spaces], ['acme/space']) self.assertEqual(request.call_count, 2) def test_huggingface_retry_config_reaches_discovery(self): config = console_runner.load_config(str(APP_DIR / 'config.yaml')) args = console_runner.build_args_from_source_config( 'huggingface', config['sources']['huggingface'], config['global'], 'spaces', ) with mock.patch.object( console_runner, 'fetch_huggingface_spaces', return_value=[], ) as fetch: self.assertEqual(console_runner.fetch_targets(args), []) self.assertEqual(args.huggingface_discovery_request_attempts, 3) self.assertEqual(args.huggingface_discovery_retry_delay, 5) self.assertEqual(fetch.call_args.kwargs['request_attempts'], 3) self.assertEqual(fetch.call_args.kwargs['retry_delay'], 5) @staticmethod def ci_args(limit=2, scan_limit=3): return SimpleNamespace( database_path='', database_url='', save_dir='', platform='github_actions', ci_seed_sources='github', ci_max_repos_per_cycle=limit, ci_seed_scan_limit=scan_limit, ci_soft_cooldown_days=0, ci_use_finding_seeds=False, ) @staticmethod def ci_rows(): return [ { 'id': index, 'source': 'github', 'query': 'q', 'target': f'https://github.com/acme/{name}.git', 'normalized_target': '', 'latest_seen': '2026-07-19', } for index, name in enumerate(('known', 'new-one', 'new-two', 'outside-bound'), 1) ] def ci_db(self, lookup_error=False): rows = self.ci_rows() class Connection: is_postgres = True def __init__(self): self.query = '' self.params = [] def execute(self, query, params): self.query = query self.params = list(params) if 'FROM target_scans' not in query: raise AssertionError(query) return SimpleNamespace(fetchall=lambda: rows[:int(params[-1])]) def commit(self): return None class DB: conn = Connection() def __init__(self): self.calls = [] def known_target_normalizations(self, source, platform): raise AssertionError('broad history lookup must not be called') def known_target_normalizations_for(self, source, platform, targets): self.calls.append((source, platform, list(targets))) if lookup_error: raise RuntimeError('lookup unavailable') return { console_runner.normalize_target(target, platform) for target in targets if json.loads(target).get('repo') == 'acme/known' } return DB() def test_ci_lookup_is_bounded_to_offered_batch_and_filters_known(self): db = self.ci_db() targets = console_runner.fetch_ci_repo_targets_from_db(self.ci_args(), db, 'github') self.assertIn('CROSS JOIN LATERAL', db.conn.query) self.assertIn('ended_at IS NOT NULL', db.conn.query) self.assertIn('ORDER BY ended_at DESC, id DESC', db.conn.query) self.assertEqual(db.conn.params, ['github', 3, 3]) self.assertEqual(len(db.calls), 1) self.assertEqual(db.calls[0][:2], ('github_actions', 'github_actions')) self.assertEqual(len(db.calls[0][2]), 3) self.assertEqual( [json.loads(target)['repo'] for target in targets], ['acme/new-one', 'acme/new-two'], ) def test_known_lookup_errors_fail_open_for_page_and_ci_discovery(self): def failed_lookup(_targets): raise RuntimeError('lookup unavailable') self.assertFalse(scanner.page_is_known( ['https://github.com/acme/repo.git'], normalize_target=lambda value: value, known_target_lookup=failed_lookup, )) db = self.ci_db(lookup_error=True) targets = console_runner.fetch_ci_repo_targets_from_db(self.ci_args(), db, 'github') self.assertEqual( [json.loads(target)['repo'] for target in targets], ['acme/known', 'acme/new-one'], ) def test_ci_cooldown_uses_bounded_keyset_pages_and_reports_truncation(self): cooldown_rows = [ { 'id': row_id, 'target': f'https://github.com/acme/cooldown-{row_id}.git', 'skipped_reason': 'ci_no_config', 'ended_at': '2026-07-20T00:00:00+00:00', } for row_id in range(301, 0, -1) ] class Connection: is_postgres = True def __init__(self): self.cooldown_limits = [] self.cooldown_sql = [] self.cooldown_pages = 0 self.commits = 0 self.rollbacks = 0 def execute(self, query, params): normalized = ' '.join(query.split()) if "status = 'skipped'" in normalized and 'SELECT 1' in normalized: return SimpleNamespace(fetchone=lambda: {'exists': 1}) if "status = 'skipped'" in normalized: limit = int(params[-1]) start = sum(self.cooldown_limits) self.cooldown_limits.append(limit) self.cooldown_sql.append(normalized) self.cooldown_pages += 1 return SimpleNamespace(fetchall=lambda: cooldown_rows[start:start + limit]) if 'CROSS JOIN LATERAL' in normalized: return SimpleNamespace(fetchall=lambda: []) raise AssertionError(normalized) def commit(self): self.commits += 1 def rollback(self): self.rollbacks += 1 connection = Connection() db = SimpleNamespace(conn=connection) args = self.ci_args(limit=2, scan_limit=300) args.ci_soft_cooldown_days = 7 args.ci_seed_query_batch_size = 250 with mock.patch.object(console_runner, 'safe_print') as output: targets = console_runner.fetch_ci_repo_targets_from_db(args, db, 'github') self.assertEqual(targets, []) self.assertEqual(connection.cooldown_limits, [250, 50]) self.assertIn('(ended_at, id) < (?, ?)', connection.cooldown_sql[1]) self.assertEqual(connection.cooldown_pages, 2) self.assertEqual(connection.commits, 1) self.assertEqual(connection.rollbacks, 0) self.assertIn('bounded 300-row limit', output.call_args.args[0]) def test_ci_cooldown_timeout_rolls_back_postgres_transaction(self): class Connection: is_postgres = True def __init__(self): self.rollbacks = 0 def execute(self, query, params): raise TimeoutError('statement timeout') def rollback(self): self.rollbacks += 1 connection = Connection() args = self.ci_args() args.ci_soft_cooldown_days = 7 with self.assertRaisesRegex(TimeoutError, 'statement timeout'): console_runner.fetch_ci_repo_targets_from_db( args, SimpleNamespace(conn=connection), 'github', ) self.assertEqual(connection.rollbacks, 1) def test_ci_seed_statement_timeout_discards_partial_rows_and_keeps_connection_usable(self): class QueryCanceled(Exception): sqlstate = '57014' class Connection: is_postgres = True def __init__(self): self.pages = 0 self.rollbacks = 0 self.poisoned = False def execute(self, query, params): normalized = ' '.join(query.split()) if normalized == 'SELECT 1': if self.poisoned: raise RuntimeError('connection remained poisoned') return SimpleNamespace(fetchone=lambda: {'ok': 1}) if 'CROSS JOIN LATERAL' not in normalized: raise AssertionError(normalized) self.pages += 1 if self.pages == 2: self.poisoned = True raise QueryCanceled('canceling statement due to statement timeout') rows = [ { 'id': index, 'source': 'github', 'query': 'stale', 'target': f'https://github.com/acme/stale-{index}.git', 'normalized_target': '', 'latest_seen': '2026-07-20', } for index in range(25) ] return SimpleNamespace(fetchall=lambda: rows) def rollback(self): self.poisoned = False self.rollbacks += 1 class DB: def __init__(self, connection): self.conn = connection self.lookup_calls = 0 def known_target_normalizations_for(self, *args): self.lookup_calls += 1 raise AssertionError('partial seed rows must not be used') connection = Connection() db = DB(connection) args = self.ci_args(limit=2, scan_limit=30) args.ci_seed_query_batch_size = 25 with mock.patch.object(console_runner, 'safe_print') as output: targets = console_runner.fetch_ci_repo_targets_from_db(args, db, 'github') self.assertEqual(targets, []) self.assertEqual(connection.pages, 2) self.assertEqual(connection.rollbacks, 1) self.assertEqual(db.lookup_calls, 0) self.assertEqual(connection.execute('SELECT 1', ()).fetchone()['ok'], 1) self.assertIn('no seed targets', output.call_args.args[0]) self.assertNotIn('stale-0', output.call_args.args[0]) def test_ci_seed_non_timeout_database_defect_remains_fail_closed(self): class Connection: is_postgres = True def __init__(self): self.rollbacks = 0 def execute(self, query, params): raise RuntimeError('programming defect') def rollback(self): self.rollbacks += 1 connection = Connection() with self.assertRaisesRegex(RuntimeError, 'programming defect'): console_runner.fetch_ci_repo_targets_from_db( self.ci_args(), SimpleNamespace(conn=connection), 'github', ) self.assertEqual(connection.rollbacks, 0) def test_ci_seed_timeout_without_confirmed_rollback_remains_fail_closed(self): class Connection: is_postgres = True def execute(self, query, params): raise TimeoutError('statement timeout') def rollback(self): raise OSError('rollback unavailable') db = SimpleNamespace(conn=Connection()) with self.assertRaisesRegex(TimeoutError, 'statement timeout'): console_runner.fetch_ci_repo_targets_from_db( self.ci_args(), db, 'github', ) def test_source_failure_rolls_back_before_cycle_finalization(self): order = [] class Connection: is_postgres = True def rollback(self): order.append('rollback') class DB: conn = Connection() def start_source_cycle(self, *args, **kwargs): return 7 def finish_source_cycle(self, *args, **kwargs): order.append('finish') config = { 'global': {}, 'sources': {'github': {'queries': ['fixture'], 'mode': 'recent'}}, } state = {'sources': {'github': console_runner.default_source_state()}} args = SimpleNamespace(platform='github', mode='recent') with mock.patch.object(console_runner, 'save_state'), \ mock.patch.object(console_runner, 'select_auth_entry', return_value=None), \ mock.patch.object(console_runner, 'refresh_auth_summary'), \ mock.patch.object(console_runner, 'build_args_from_source_config', return_value=args), \ mock.patch.object(console_runner, 'configure_source_auth'), \ mock.patch.object(console_runner, 'queue_counts_for_args', return_value={}), \ mock.patch.object(console_runner, 'run_cycle', side_effect=RuntimeError('statement timeout')): with self.assertRaisesRegex(RuntimeError, 'statement timeout'): console_runner.run_configured_source( 'github', config, state, 'state.json', {}, DB(), run_id=3, ) self.assertEqual(order, ['rollback', 'finish']) def test_backlog_only_cycle_does_not_advance_discovery_query(self): class Connection: is_postgres = True class DB: conn = Connection() def start_source_cycle(self, *args, **kwargs): return 7 config = { 'global': {}, 'sources': {'github': {'queries': ['first', 'second'], 'mode': 'recent'}}, } state = {'sources': {'github': console_runner.default_source_state()}} args = SimpleNamespace(platform='github', mode='recent') with mock.patch.object(console_runner, 'save_state'), \ mock.patch.object(console_runner, 'select_auth_entry', return_value=None), \ mock.patch.object(console_runner, 'refresh_auth_summary'), \ mock.patch.object(console_runner, 'build_args_from_source_config', return_value=args), \ mock.patch.object(console_runner, 'configure_source_auth'), \ mock.patch.object(console_runner, 'queue_counts_for_args', return_value={}), \ mock.patch.object(console_runner, 'run_cycle', return_value={ 'scanned_count': 3, 'source_failure_count': 0, 'backlog_only': True, }): metrics = console_runner.run_configured_source( 'github', config, state, 'state.json', {}, DB(), run_id=3, ) self.assertTrue(metrics['backlog_only']) self.assertEqual(state['sources']['github']['query_index'], 0) def test_configured_source_refreshes_discovery_while_backlog_exists(self): class DB: url = 'postgresql://fixture' last_error = '' @staticmethod def require_runtime_safety_schema(): return True @staticmethod def require_final_cutover(): return True @staticmethod def has_claimable_targets_v2(*args, **kwargs): return True @staticmethod def finish_source_cycle(*args, **kwargs): return True class Lease: releasable = True def __init__(self): self.released = False def release(self): self.released = True args = SimpleNamespace( platform='docker', workers=1, max_targets=1, timeout=10, target_retry_max_attempts=3, refresh_registry=True, target_claim_order='balanced', result_bundle_min_free_bytes=0, result_bundle_max_event_bytes=1024 * 1024, projection_backlog_max_items=10, projection_backlog_max_bytes=8 * 1024 * 1024, projection_backlog_headroom_bytes=2 * 1024 * 1024, result_spool_wait_sec=0.01, ) identity = SimpleNamespace(as_dict=lambda: { 'pid': 1, 'creation_time': 'fixture', 'executable': 'python', }) lease = Lease() with tempfile.TemporaryDirectory() as temp_dir, \ mock.patch.dict(os.environ, {'TRUF_SUPERVISOR_INSTANCE_ID': 'fixture'}), \ mock.patch.object(scanner.scan_config, 'max_active_scans', 1), \ mock.patch.object(console_runner, 'require_private_directory', return_value=temp_dir), \ mock.patch.object(console_runner, 'current_process_identity', return_value=identity), \ mock.patch.object(console_runner, 'queue_files_for_args', return_value=(None, None)), \ mock.patch.object(console_runner, 'fetch_targets', return_value=['fresh-target']) as fetch, \ mock.patch.object(console_runner, 'prepare_targets', return_value=( [], None, None, {'fetched_count': 1, 'queued_new_count': 1}, )) as prepare, \ mock.patch.object(console_runner, 'prepare_scan_options', return_value={}), \ mock.patch.object(console_runner, 'acquire_scan_slot', return_value=lease), \ mock.patch.object(console_runner, 'reserve_v2_admission_with_recovery', return_value= console_runner.V2AdmissionOutcome(None, False)), \ mock.patch.object(console_runner, 'supervised_spool_stop_requested', return_value=False): metrics = console_runner.run_cycle_v2(args, DB(), 1, 2, 'dockerhub') fetch.assert_called_once() self.assertTrue(prepare.call_args.kwargs['enqueue_only']) self.assertFalse(metrics['backlog_only']) self.assertEqual(metrics['queued_new_count'], 1) self.assertTrue(lease.released) def test_source_process_private_bytes_is_telemetry_only(self): self.assertNotIn('source_process_recycle_due', dir(console_runner)) self.assertNotIn('source_process_memory_limit_exceeded', dir(console_runner)) self.assertGreaterEqual(console_runner.current_process_private_bytes(), 0) if __name__ == '__main__': unittest.main()