import json import io import os import shutil import sqlite3 import tempfile import unittest from pathlib import Path from web.config import Config class ApplicationIntegrationTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.temp_dir = tempfile.TemporaryDirectory() cls.original_web_path = Config.DB_WEB_PATH Config.DB_WEB_PATH = os.path.join(cls.temp_dir.name, 'Web_App.sqlite') shutil.copy2(cls.original_web_path, Config.DB_WEB_PATH) from web.app import create_app cls.app = create_app() cls.app.config.update(TESTING=True) cls.client = cls.app.test_client() with sqlite3.connect(Config.DB_WEB_PATH) as db: raw_ids = db.execute( """ SELECT player_ids_json FROM team_lineups WHERE is_active = 1 LIMIT 1 """ ).fetchone()[0] cls.roster_ids = [str(value) for value in json.loads(raw_ids)] @classmethod def tearDownClass(cls): Config.DB_WEB_PATH = cls.original_web_path cls.temp_dir.cleanup() def test_primary_pages_render(self): paths = [ '/', '/matches/', '/players/', f'/players/{self.roster_ids[0]}', '/teams/', '/tactics/', '/opponents/', ] for path in paths: with self.subTest(path=path): response = self.client.get(path) self.assertEqual(response.status_code, 200) self.assertGreater(len(response.data), 100) def test_admin_integrity_page_and_json_render(self): with self.client.session_transaction() as session: session['is_admin'] = True response = self.client.get('/admin/data-integrity') self.assertEqual(response.status_code, 200) self.assertIn('数据完整性中心'.encode('utf-8'), response.data) response = self.client.get('/admin/data-integrity?format=json') self.assertEqual(response.status_code, 200) report = response.get_json() self.assertIn(report['overall_status'], {'pass', 'warn', 'fail'}) self.assertGreater(report['counts']['matches'], 0) self.assertEqual( report['counts']['web_schema_version'], Config.WEB_SCHEMA_VERSION, ) response = self.client.get('/admin/import-match') self.assertEqual(response.status_code, 200) self.assertIn('比赛数据导入'.encode('utf-8'), response.data) def test_duplicate_match_upload_is_rejected_without_starting_job(self): from database.paths import L1_DB with self.client.session_transaction() as session: session['is_admin'] = True with sqlite3.connect(str(L1_DB)) as db: raw = db.execute( """ SELECT content FROM raw_iframe_network ORDER BY match_id LIMIT 1 """ ).fetchone()[0] response = self.client.post( '/admin/import-match', data={ 'capture': ( io.BytesIO(raw.encode('utf-8')), 'iframe_network.json', ), }, content_type='multipart/form-data', follow_redirects=True, ) self.assertEqual(response.status_code, 200) self.assertIn(b'already imported with identical data', response.data) def test_player_search_works_across_l2_and_l3(self): response = self.client.get('/players/?search=jAck') self.assertEqual(response.status_code, 200) self.assertIn(b'jAckY0987', response.data) def test_profile_keeps_all_primary_sections(self): response = self.client.get(f'/players/{self.roster_ids[0]}') self.assertEqual(response.status_code, 200) for label in ( '近期表现走势', '能力八维图', 'CORE (核心表现)', '阶段表现', '职业纪录', '比赛记录', '地图数据', '留言板', ): with self.subTest(label=label): self.assertIn(label.encode('utf-8'), response.data) def test_profile_period_api_and_trend_window(self): steam_id = self.roster_ids[0] response = self.client.get( f'/players/{steam_id}/period_stats?period=last_20' ) self.assertEqual(response.status_code, 200) period = response.get_json() self.assertEqual(period['period_key'], 'last_20') self.assertEqual(period['matches'], 20) self.assertEqual(period['sample_reliable'], 1) response = self.client.get( f'/players/{steam_id}/charts_data?period=last_10' ) self.assertEqual(response.status_code, 200) chart = response.get_json() self.assertEqual(chart['period']['period_key'], 'last_10') self.assertLessEqual(len(chart['trend']['labels']), 10) def test_date_filter_uses_unix_timestamp_conversion(self): from web.services.stats_service import StatsService with self.app.app_context(): matches, total = StatsService.get_matches( page=1, per_page=20, date_from='2025-01-01', date_to='2026-12-31', ) self.assertGreater(total, 0) self.assertGreater(len(matches), 0) def test_shared_matches_require_same_team(self): from web.services.stats_service import StatsService selected_ids = self.roster_ids[:2] with self.app.app_context(): matches = StatsService.get_shared_matches(selected_ids) self.assertGreater(len(matches), 0) l2 = sqlite3.connect(Config.DB_L2_PATH) try: for match in matches: placeholders = ','.join('?' for _ in selected_ids) team_count = l2.execute( f""" SELECT COUNT(DISTINCT team_id) FROM fact_match_players WHERE match_id = ? AND steam_id_64 IN ({placeholders}) """, [match['match_id']] + selected_ids, ).fetchone()[0] self.assertEqual(team_count, 1) finally: l2.close() def test_opponent_list_only_contains_actual_opponents(self): from web.services.opponent_service import OpponentService with self.app.app_context(): opponents, total = OpponentService.get_opponent_list( page=1, per_page=20, ) self.assertGreater(total, 0) self.assertGreater(len(opponents), 0) l2 = sqlite3.connect(Config.DB_L2_PATH) try: roster_ph = ','.join('?' for _ in self.roster_ids) for opponent in opponents: faced = l2.execute( f""" SELECT COUNT(*) FROM fact_match_players opponent JOIN fact_match_players roster ON roster.match_id = opponent.match_id AND roster.team_id != opponent.team_id WHERE opponent.steam_id_64 = ? AND roster.steam_id_64 IN ({roster_ph}) """, [opponent['steam_id_64']] + self.roster_ids, ).fetchone()[0] self.assertGreater(faced, 0) finally: l2.close() class L3MartBuilderTests(unittest.TestCase): def test_auxiliary_marts_build_on_database_copy(self): from database.L3.L3_Builder import ( _get_team_players, _rebuild_auxiliary_marts, ) with tempfile.TemporaryDirectory() as temp_dir: l3_path = os.path.join(temp_dir, 'L3.db') shutil.copy2(Config.DB_L3_PATH, l3_path) l2 = sqlite3.connect(Config.DB_L2_PATH) l2.row_factory = sqlite3.Row l3 = sqlite3.connect(l3_path) try: _rebuild_auxiliary_marts( l2, l3, sorted(_get_team_players()), ) l3.commit() for table in ( 'dm_player_match_history', 'dm_player_map_stats', 'dm_player_period_stats', 'dm_player_records', 'dm_player_weapon_stats', ): count = l3.execute( f'SELECT COUNT(*) FROM {table}' ).fetchone()[0] self.assertGreater(count, 0) self.assertEqual( l3.execute('PRAGMA quick_check').fetchone()[0], 'ok', ) finally: l2.close() l3.close() def test_spatial_processor_does_not_emit_fake_geometry_metrics(self): from database.L3.L3_Builder import _get_team_players from database.L3.processors.intelligence_processor import IntelligenceProcessor roster_ids = sorted(_get_team_players()) placeholders = ','.join('?' for _ in roster_ids) l2 = sqlite3.connect(Config.DB_L2_PATH) try: row = l2.execute( f""" SELECT attacker_steam_id FROM fact_round_events WHERE attacker_steam_id IN ({placeholders}) AND attacker_pos_x IS NOT NULL GROUP BY attacker_steam_id ORDER BY COUNT(*) DESC LIMIT 1 """, roster_ids, ).fetchone() self.assertIsNotNone(row) features = IntelligenceProcessor._calculate_position_mastery( str(row[0]), l2, ) finally: l2.close() for key in ( 'int_pos_site_a_control_rate', 'int_pos_site_b_control_rate', 'int_pos_mid_control_rate', 'int_pos_rotation_speed', 'int_pos_lurk_tendency', 'int_pos_site_anchor_score', 'int_pos_retake_positioning', 'int_pos_postplant_positioning', 'int_pos_avg_distance_from_teammates', ): with self.subTest(key=key): self.assertIsNone(features[key]) self.assertIsNotNone(features['int_pos_position_diversity']) def test_percentiles_are_calculated_from_peer_scores(self): from database.L3.L3_Builder import _update_percentiles with tempfile.TemporaryDirectory() as temp_dir: l3_path = os.path.join(temp_dir, 'L3.db') shutil.copy2(Config.DB_L3_PATH, l3_path) l3 = sqlite3.connect(l3_path) try: player_ids = [ row[0] for row in l3.execute( """ SELECT steam_id_64 FROM dm_player_features ORDER BY steam_id_64 LIMIT 3 """ ) ] for steam_id, score in zip(player_ids, (10.0, 20.0, 30.0)): l3.execute( """ UPDATE dm_player_features SET score_overall = ? WHERE steam_id_64 = ? """, [score, steam_id], ) _update_percentiles(l3, player_ids) values = [ row[0] for row in l3.execute( f""" SELECT tier_percentile FROM dm_player_features WHERE steam_id_64 IN ({','.join('?' for _ in player_ids)}) ORDER BY score_overall """, player_ids, ) ] self.assertEqual(values, [33.33, 66.67, 100.0]) finally: l3.close() def test_l3_backup_is_a_valid_sqlite_database(self): from database.L3.L3_Builder import _backup_l3_database with tempfile.TemporaryDirectory() as temp_dir: source = os.path.join(temp_dir, 'source.db') backup = os.path.join(temp_dir, 'backup.db') shutil.copy2(Config.DB_L3_PATH, source) result_path = _backup_l3_database(source, backup) self.assertEqual(result_path, backup) with sqlite3.connect(backup) as db: self.assertEqual(db.execute('PRAGMA quick_check').fetchone()[0], 'ok') class FeatureFormulaTests(unittest.TestCase): def test_empty_pace_inputs_do_not_receive_free_points(self): from database.L3.processors.composite_processor import CompositeProcessor self.assertEqual(CompositeProcessor._calculate_pace_score({}), 0.0) def test_lower_map_and_elo_volatility_improves_stability_score(self): from database.L3.processors.composite_processor import CompositeProcessor common = { 'meta_rating_volatility': 0.2, 'meta_loss_rating': 1.0, 'meta_rating_consistency': 70, 'int_pressure_tilt_resistance': 0.8, 'meta_recent_form_rating': 1.15, } stable = dict(common, meta_map_stability=0.05, meta_elo_tier_stability=0.05) volatile = dict(common, meta_map_stability=0.25, meta_elo_tier_stability=0.48) self.assertGreater( CompositeProcessor._calculate_stability_score(stable), CompositeProcessor._calculate_stability_score(volatile), ) def test_recent_form_uses_match_time_and_elo_stability_is_calculated(self): from database.L3.L3_Builder import _get_team_players from database.L3.processors.meta_processor import MetaProcessor steam_id = sorted(_get_team_players())[0] l2 = sqlite3.connect(Config.DB_L2_PATH) try: expected_rows = l2.execute( """ SELECT p.rating FROM fact_match_players p JOIN fact_matches m ON m.match_id = p.match_id WHERE p.steam_id_64 = ? ORDER BY m.start_time DESC, p.match_id DESC LIMIT 10 """, [steam_id], ).fetchall() expected = sum(row[0] for row in expected_rows) / len(expected_rows) features = MetaProcessor._calculate_stability(steam_id, l2) finally: l2.close() self.assertAlmostEqual( features['meta_recent_form_rating'], round(expected, 3), places=3, ) self.assertGreaterEqual(features['meta_elo_tier_stability'], 0) self.assertNotEqual( features['meta_elo_tier_stability'], features['meta_rating_volatility'], ) class DatabaseGovernanceTests(unittest.TestCase): def test_database_paths_are_absolute_and_exist(self): from database.paths import L1_DB, L2_DB, L3_DB, WEB_DB for path in (L1_DB, L2_DB, L3_DB, WEB_DB): with self.subTest(path=path): self.assertTrue(path.is_absolute()) self.assertTrue(path.exists()) def test_valid_capture_is_identified_from_network_urls(self): from database.paths import L1_DB from web.services.import_service import MatchImportService with sqlite3.connect(str(L1_DB)) as db: match_id, raw = db.execute( """ SELECT match_id, content FROM raw_iframe_network ORDER BY match_id LIMIT 1 """ ).fetchone() result = MatchImportService.validate_capture(raw.encode('utf-8')) self.assertEqual(result['match_id'], match_id) self.assertGreaterEqual(result['successful_responses'], 2) def test_prepare_import_is_atomic_and_rejects_duplicate_queue(self): import web.services.import_service as import_module from database.paths import L1_DB from web.services.import_service import ( DuplicateMatchError, MatchImportService, ) with sqlite3.connect(str(L1_DB)) as db: original_id, raw = db.execute( """ SELECT match_id, content FROM raw_iframe_network ORDER BY match_id LIMIT 1 """ ).fetchone() new_id = 'g161-99999999999999999999999' raw = raw.replace(original_id, new_id).encode('utf-8') with tempfile.TemporaryDirectory() as temp_dir: old_l1 = import_module.L1_DB old_arena = import_module.OUTPUT_ARENA old_web = Config.DB_WEB_PATH fake_l1 = Path(temp_dir) / 'L1.db' fake_web = Path(temp_dir) / 'Web_App.sqlite' with sqlite3.connect(str(fake_l1)) as db: db.execute( """ CREATE TABLE raw_iframe_network ( match_id TEXT PRIMARY KEY, content TEXT ) """ ) with sqlite3.connect(str(fake_web)) as db: with open(Config.DB_WEB_SCHEMA_PATH, 'r', encoding='utf-8') as schema: db.executescript(schema.read()) import_module.L1_DB = fake_l1 import_module.OUTPUT_ARENA = Path(temp_dir) / 'output_arena' Config.DB_WEB_PATH = str(fake_web) try: prepared = MatchImportService.prepare_import( raw, 'iframe_network.json', created_by='test', ) self.assertEqual(prepared['match_id'], new_id) self.assertTrue(Path(prepared['source_path']).exists()) with self.assertRaises(DuplicateMatchError): MatchImportService.prepare_import( raw, 'iframe_network.json', created_by='test', ) finally: import_module.L1_DB = old_l1 import_module.OUTPUT_ARENA = old_arena Config.DB_WEB_PATH = old_web def test_pipeline_post_validation_accepts_current_databases(self): from database.paths import L1_DB from database.pipeline import _validate_pipeline_output with sqlite3.connect(str(L1_DB)) as db: match_id = db.execute( 'SELECT match_id FROM raw_iframe_network ORDER BY match_id LIMIT 1' ).fetchone()[0] _validate_pipeline_output(match_id) def test_job_store_tracks_progress_logs_and_completion(self): from database.job_store import JobStore with tempfile.TemporaryDirectory() as temp_dir: web_path = Path(temp_dir) / 'Web.sqlite' with sqlite3.connect(str(web_path)) as db: with open(Config.DB_WEB_SCHEMA_PATH, 'r', encoding='utf-8') as schema: db.executescript(schema.read()) store = JobStore(web_path) job_id = store.create_job( 'test_pipeline', match_id='g161-99999999999999999999999', created_by='test', ) store.start_job(job_id, 'backup', 'Starting') store.update_progress(job_id, 'l2', 55, 'Building L2') store.append_log(job_id, 'line one\n') store.finish_job(job_id, True, 'Done', 1.25) job = store.get_job(job_id) self.assertEqual(job['status'], 'succeeded') self.assertEqual(job['progress'], 100) self.assertEqual(job['current_stage'], 'complete') self.assertIn('line one', job['log_text']) self.assertEqual(job['duration_seconds'], 1.25) def test_database_backup_and_restore_round_trip(self): from database.maintenance import backup_database, restore_database with tempfile.TemporaryDirectory() as temp_dir: source = Path(temp_dir) / 'source.db' backup = Path(temp_dir) / 'backup.db' with sqlite3.connect(str(source)) as db: db.execute('CREATE TABLE values_table (value INTEGER)') db.execute('INSERT INTO values_table VALUES (1)') backup_database(source, backup) with sqlite3.connect(str(source)) as db: db.execute('UPDATE values_table SET value = 2') restore_database(backup, source) with sqlite3.connect(str(source)) as db: value = db.execute( 'SELECT value FROM values_table' ).fetchone()[0] self.assertEqual(value, 1) def test_high_frequency_queries_use_operational_indexes(self): l2 = sqlite3.connect(Config.DB_L2_PATH) try: plans = { 'player': l2.execute( """ EXPLAIN QUERY PLAN SELECT * FROM fact_match_players WHERE steam_id_64 = ? """, ['76561198330488905'], ).fetchone()[3], 'party': l2.execute( """ EXPLAIN QUERY PLAN SELECT * FROM fact_match_players WHERE match_id = ? AND match_team_id = ? """, ['match', 1], ).fetchone()[3], 'victim': l2.execute( """ EXPLAIN QUERY PLAN SELECT * FROM fact_round_events WHERE victim_steam_id = ? """, ['player'], ).fetchone()[3], } finally: l2.close() self.assertIn('idx_match_players_player_match', plans['player']) self.assertIn('idx_match_players_party', plans['party']) self.assertIn('idx_round_events_victim', plans['victim']) def test_player_records_reference_real_matches(self): l3 = sqlite3.connect(Config.DB_L3_PATH) l2 = sqlite3.connect(Config.DB_L2_PATH) try: records = l3.execute( """ SELECT match_id FROM dm_player_records WHERE match_id IS NOT NULL """ ).fetchall() self.assertGreater(len(records), 0) for (match_id,) in records: exists = l2.execute( 'SELECT 1 FROM fact_matches WHERE match_id = ?', [match_id], ).fetchone() self.assertIsNotNone(exists) finally: l3.close() l2.close() if __name__ == '__main__': unittest.main()