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/', '/reports/', '/awards/', '/players/', f'/players/{self.roster_ids[0]}', '/teams/', '/teams/performance', '/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_match_detail_contains_post_match_report(self): l3 = sqlite3.connect(Config.DB_L3_PATH) try: match_id = l3.execute( """ SELECT match_id FROM dm_match_reports ORDER BY match_date DESC LIMIT 1 """ ).fetchone()[0] finally: l3.close() response = self.client.get(f'/matches/{match_id}') self.assertEqual(response.status_code, 200) self.assertIn('赛后报告'.encode('utf-8'), response.data) self.assertIn('Form Improver'.encode('utf-8'), response.data) def test_awards_and_professional_identity_render(self): response = self.client.get('/awards/?type=monthly') self.assertEqual(response.status_code, 200) self.assertIn('周期最佳与荣誉榜'.encode('utf-8'), response.data) self.assertIn('月度最佳'.encode('utf-8'), response.data) response = self.client.get(f'/players/{self.roster_ids[0]}') self.assertEqual(response.status_code, 200) self.assertIn(b'Professional Identity', response.data) self.assertIn('代表武器'.encode('utf-8'), response.data) def test_team_performance_page_contains_beta_sections(self): response = self.client.get('/teams/performance') self.assertEqual(response.status_code, 200) for label in ( '战队履历与阵容表现', '二人组表现', '实际阵容组合', 'Roster Version 历史', ): with self.subTest(label=label): self.assertIn(label.encode('utf-8'), response.data) 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] upload = io.BytesIO(raw.encode('utf-8')) try: response = self.client.post( '/admin/import-match', data={ 'capture': ( upload, 'iframe_network.json', ), }, content_type='multipart/form-data', follow_redirects=True, ) finally: upload.close() try: self.assertEqual(response.status_code, 200) self.assertIn(b'already imported with identical data', response.data) finally: response.close() 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_duo_stats', 'dm_lineup_stats', 'dm_match_player_reports', 'dm_match_reports', 'dm_player_awards', 'dm_player_match_history', 'dm_player_map_stats', 'dm_player_period_stats', 'dm_player_record_events', 'dm_player_records', 'dm_player_weapon_stats', 'dm_team_season_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() def test_award_minimum_samples_and_unique_period_winner(self): minimums = { 'daily': 1, 'weekly': 2, 'monthly': 5, 'quarterly': 10, 'yearly': 20, } l3 = sqlite3.connect(Config.DB_L3_PATH) try: rows = l3.execute( """ SELECT award_type, period_key, matches FROM dm_player_awards """ ).fetchall() finally: l3.close() self.assertGreater(len(rows), 0) seen = set() for award_type, period_key, matches in rows: with self.subTest(award_type=award_type, period_key=period_key): self.assertGreaterEqual(matches, minimums[award_type]) self.assertNotIn((award_type, period_key), seen) seen.add((award_type, period_key)) def test_seasons_only_include_team_matches(self): l2 = sqlite3.connect(Config.DB_L2_PATH) l3 = sqlite3.connect(Config.DB_L3_PATH) web = sqlite3.connect(Config.DB_WEB_PATH) try: roster_ids = json.loads( web.execute( """ SELECT player_ids_json FROM team_lineups WHERE is_active = 1 """ ).fetchone()[0] ) placeholders = ','.join('?' for _ in roster_ids) expected = { str(row[0]): row[1] for row in l2.execute( f""" SELECT strftime('%Y', m.start_time, 'unixepoch') AS season, COUNT(DISTINCT grouped.match_id) FROM ( SELECT match_id, CASE WHEN group_id IN (1, 2) THEN group_id WHEN team_id IN (1, 2) THEN team_id END AS team_key FROM fact_match_players WHERE steam_id_64 IN ({placeholders}) GROUP BY match_id, team_key HAVING COUNT(DISTINCT steam_id_64) >= 2 AND team_key IS NOT NULL ) grouped JOIN fact_matches m ON m.match_id = grouped.match_id GROUP BY strftime('%Y', m.start_time, 'unixepoch') """, roster_ids, ) } actual = { row[0]: row[1] for row in l3.execute( 'SELECT season_key, matches FROM dm_team_season_stats' ) } finally: l2.close() l3.close() web.close() self.assertEqual(actual, expected) def test_roster_version_snapshot_closes_previous_version(self): from web.services.roster_version_service import RosterVersionService with tempfile.TemporaryDirectory() as temp_dir: web_path = Path(temp_dir) / 'Web.sqlite' l2_path = Path(temp_dir) / 'L2.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()) db.execute( """ INSERT INTO team_lineups ( name, player_ids_json, is_active ) VALUES ('Test', '["1", "2"]', 1) """ ) with sqlite3.connect(str(l2_path)) as db: db.executescript( """ CREATE TABLE fact_matches ( match_id TEXT PRIMARY KEY, start_time INTEGER ); CREATE TABLE fact_match_players ( match_id TEXT, steam_id_64 TEXT ); INSERT INTO fact_matches VALUES ('m1', 100); INSERT INTO fact_match_players VALUES ('m1', '1'); INSERT INTO fact_match_players VALUES ('m1', '2'); """ ) first_id = RosterVersionService.ensure_initial_version( str(web_path), str(l2_path), ) second_id, created = RosterVersionService.snapshot_roster( ['1', '3'], effective_from=200, web_db_path=str(web_path), ) RosterVersionService.update_current_member_role( '1', 'starter', web_db_path=str(web_path), ) self.assertTrue(created) self.assertNotEqual(first_id, second_id) with sqlite3.connect(str(web_path)) as db: old = db.execute( """ SELECT is_current, effective_to FROM team_roster_versions WHERE id = ? """, [first_id], ).fetchone() current_ids = [ row[0] for row in db.execute( """ SELECT steam_id_64 FROM team_roster_members WHERE roster_version_id = ? ORDER BY position_order """, [second_id], ) ] role = db.execute( """ SELECT member_role FROM team_roster_members WHERE roster_version_id = ? AND steam_id_64 = '1' """, [second_id], ).fetchone()[0] self.assertEqual(old, (0, 199)) self.assertEqual(current_ids, ['1', '3']) self.assertEqual(role, 'starter') def test_team_marts_cover_active_roster_combinations(self): import json from math import comb web = sqlite3.connect(Config.DB_WEB_PATH) l3 = sqlite3.connect(Config.DB_L3_PATH) try: roster_ids = json.loads( web.execute( """ SELECT player_ids_json FROM team_lineups WHERE is_active = 1 """ ).fetchone()[0] ) duo_count = l3.execute( 'SELECT COUNT(*) FROM dm_duo_stats' ).fetchone()[0] lineup_sizes = { row[0]: row[1] for row in l3.execute( """ SELECT player_count, COUNT(*) FROM dm_lineup_stats GROUP BY player_count """ ) } finally: web.close() l3.close() self.assertEqual(duo_count, comb(len(roster_ids), 2)) self.assertTrue(all(size in lineup_sizes for size in (2, 3, 4, 5))) if __name__ == '__main__': unittest.main()