1141 lines
42 KiB
Python
1141 lines
42 KiB
Python
import json
|
|
import io
|
|
import os
|
|
import shutil
|
|
import sqlite3
|
|
import subprocess
|
|
import sys
|
|
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/',
|
|
'/discover/',
|
|
'/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_health_endpoint_and_brand_watermark(self):
|
|
response = self.client.get('/healthz')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.get_json()['brand'], 'Superjacky6')
|
|
response = self.client.get('/')
|
|
self.assertIn(b'Superjacky6', response.data)
|
|
self.assertIn(b'jacky', response.data)
|
|
|
|
def test_private_visibility_requires_viewer_token(self):
|
|
from web.app import create_app
|
|
|
|
class PrivateConfig(Config):
|
|
TESTING = True
|
|
SECRET_KEY = 'private-test'
|
|
SITE_VISIBILITY = 'private'
|
|
VIEWER_TOKEN = 'viewer-secret'
|
|
|
|
app = create_app(PrivateConfig)
|
|
client = app.test_client()
|
|
response = client.get('/')
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertIn('/access/', response.location)
|
|
self.assertEqual(client.get('/healthz').status_code, 200)
|
|
|
|
with client.session_transaction() as session:
|
|
session['_csrf_token'] = 'viewer-csrf'
|
|
response = client.post(
|
|
'/access/',
|
|
data={
|
|
'_csrf_token': 'viewer-csrf',
|
|
'token': 'viewer-secret',
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 302)
|
|
self.assertEqual(client.get('/').status_code, 200)
|
|
|
|
def test_discovery_filters_and_player_medals_render(self):
|
|
response = self.client.get('/discover/')
|
|
self.assertEqual(response.status_code, 200)
|
|
for label in ('趣味发现', '表现勋章墙', '勋章总榜'):
|
|
self.assertIn(label.encode('utf-8'), response.data)
|
|
|
|
response = self.client.get('/discover/?tone=negative')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('低谷收藏家'.encode('utf-8'), response.data)
|
|
|
|
response = self.client.get('/discover/?dimension=elo')
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b'ELO Medal', response.data)
|
|
|
|
l3 = sqlite3.connect(Config.DB_L3_PATH)
|
|
try:
|
|
medal_player = l3.execute(
|
|
"""
|
|
SELECT steam_id_64
|
|
FROM dm_performance_medals
|
|
ORDER BY medal_rank
|
|
LIMIT 1
|
|
"""
|
|
).fetchone()[0]
|
|
finally:
|
|
l3.close()
|
|
response = self.client.get(f'/players/{medal_player}')
|
|
self.assertIn(b'Map & ELO Medals', response.data)
|
|
|
|
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
|
|
session['_csrf_token'] = 'test-csrf'
|
|
|
|
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_admin_console_pages_share_navigation_and_real_data(self):
|
|
with self.client.session_transaction() as session:
|
|
session['is_admin'] = True
|
|
session['_csrf_token'] = 'test-csrf'
|
|
pages = {
|
|
'/admin/': ('运营总览', '完整数据流水线'),
|
|
'/admin/jobs': ('作业中心', '执行记录'),
|
|
'/admin/sql': ('只读查询控制台', 'Database Objects'),
|
|
'/admin/system': ('系统与存储', '回滚快照'),
|
|
}
|
|
for path, labels in pages.items():
|
|
with self.subTest(path=path):
|
|
response = self.client.get(path)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn('YRTV'.encode('utf-8'), response.data)
|
|
for label in labels:
|
|
self.assertIn(label.encode('utf-8'), response.data)
|
|
|
|
def test_admin_sql_is_read_only_and_csrf_protected(self):
|
|
with self.client.session_transaction() as session:
|
|
session['is_admin'] = True
|
|
session['_csrf_token'] = 'test-csrf'
|
|
|
|
response = self.client.post(
|
|
'/admin/sql',
|
|
data={
|
|
'_csrf_token': 'test-csrf',
|
|
'db_name': 'l2',
|
|
'query': 'SELECT COUNT(*) AS matches FROM fact_matches',
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b'matches', response.data)
|
|
self.assertIn(b'208', response.data)
|
|
|
|
response = self.client.post(
|
|
'/admin/sql',
|
|
data={
|
|
'_csrf_token': 'test-csrf',
|
|
'db_name': 'l2',
|
|
'query': 'DELETE FROM fact_matches',
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b'Only SELECT queries are allowed', response.data)
|
|
|
|
response = self.client.post(
|
|
'/admin/trigger_etl',
|
|
headers={'Accept': 'application/json'},
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
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
|
|
session['_csrf_token'] = 'test-csrf'
|
|
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={
|
|
'_csrf_token': 'test-csrf',
|
|
'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)
|
|
self.assertIn(b'x-data="{ selected: \'last_20\' }"', response.data)
|
|
self.assertNotIn(b'periods: [{', 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_external_data_root_bootstraps_without_touching_current_data(self):
|
|
current_hashes = {}
|
|
for path in (
|
|
Path(Config.DB_L2_PATH),
|
|
Path(Config.DB_L3_PATH),
|
|
Path(Config.DB_WEB_PATH),
|
|
):
|
|
current_hashes[str(path)] = path.stat().st_size
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env = os.environ.copy()
|
|
env['YRTV_DATA_DIR'] = temp_dir
|
|
result = subprocess.run(
|
|
[sys.executable, '-m', 'database.bootstrap'],
|
|
cwd=Config.BASE_DIR,
|
|
env=env,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
runtime = Path(temp_dir)
|
|
expected_tables = {
|
|
runtime / 'L1' / 'L1.db': 'raw_iframe_network',
|
|
runtime / 'L2' / 'L2.db': 'fact_matches',
|
|
runtime / 'L3' / 'L3.db': 'dm_player_features',
|
|
runtime / 'Web' / 'Web_App.sqlite': 'team_lineups',
|
|
}
|
|
for path, table in expected_tables.items():
|
|
with self.subTest(path=path):
|
|
self.assertTrue(path.exists())
|
|
with sqlite3.connect(str(path)) as db:
|
|
self.assertEqual(
|
|
db.execute('PRAGMA quick_check').fetchone()[0],
|
|
'ok',
|
|
)
|
|
self.assertIsNotNone(db.execute(
|
|
"""
|
|
SELECT 1 FROM sqlite_master
|
|
WHERE type = 'table' AND name = ?
|
|
""",
|
|
[table],
|
|
).fetchone())
|
|
|
|
for path, size in current_hashes.items():
|
|
self.assertEqual(Path(path).stat().st_size, size)
|
|
|
|
def test_existing_data_migrates_to_external_root(self):
|
|
from database.migrate_data import migrate_data
|
|
from database.paths import DATABASE_CODE_ROOT
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
manifest_path = migrate_data(
|
|
DATABASE_CODE_ROOT,
|
|
temp_dir,
|
|
include_imports=False,
|
|
)
|
|
manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
|
|
self.assertEqual(
|
|
set(manifest['databases']),
|
|
{'L1', 'L2', 'L3', 'Web'},
|
|
)
|
|
for item in manifest['databases'].values():
|
|
self.assertEqual(item['quick_check'], 'ok')
|
|
self.assertEqual(len(item['sha256']), 64)
|
|
|
|
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_discovery_tones_and_medal_rules(self):
|
|
l3 = sqlite3.connect(Config.DB_L3_PATH)
|
|
try:
|
|
tones = {
|
|
row[0] for row in l3.execute(
|
|
'SELECT DISTINCT tone FROM dm_discovery_insights'
|
|
)
|
|
}
|
|
medals = l3.execute(
|
|
"""
|
|
SELECT dimension_type, dimension_key, medal_rank,
|
|
medal_tier, matches
|
|
FROM dm_performance_medals
|
|
"""
|
|
).fetchall()
|
|
finally:
|
|
l3.close()
|
|
|
|
self.assertEqual(tones, {'positive', 'negative', 'fun'})
|
|
self.assertGreater(len(medals), 0)
|
|
seen = set()
|
|
expected_tiers = {1: 'gold', 2: 'silver', 3: 'bronze'}
|
|
for dimension_type, dimension_key, rank, tier, matches in medals:
|
|
key = (dimension_type, dimension_key, rank)
|
|
self.assertNotIn(key, seen)
|
|
seen.add(key)
|
|
self.assertIn(dimension_type, {'map', 'elo'})
|
|
self.assertEqual(tier, expected_tiers[rank])
|
|
self.assertGreaterEqual(matches, 5)
|
|
|
|
def test_map_gold_medal_matches_highest_eligible_rating(self):
|
|
l3 = sqlite3.connect(Config.DB_L3_PATH)
|
|
l2 = sqlite3.connect(Config.DB_L2_PATH)
|
|
web = sqlite3.connect(Config.DB_WEB_PATH)
|
|
try:
|
|
medal = l3.execute(
|
|
"""
|
|
SELECT dimension_key, steam_id_64, avg_rating
|
|
FROM dm_performance_medals
|
|
WHERE dimension_type = 'map' AND medal_rank = 1
|
|
ORDER BY dimension_key
|
|
LIMIT 1
|
|
"""
|
|
).fetchone()
|
|
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 = l2.execute(
|
|
f"""
|
|
SELECT p.steam_id_64, AVG(p.rating) AS avg_rating
|
|
FROM fact_match_players p
|
|
JOIN fact_matches m ON m.match_id = p.match_id
|
|
WHERE p.steam_id_64 IN ({placeholders})
|
|
AND m.map_name = ?
|
|
GROUP BY p.steam_id_64
|
|
HAVING COUNT(*) >= 5
|
|
ORDER BY avg_rating DESC, AVG(p.adr) DESC, COUNT(*) DESC
|
|
LIMIT 1
|
|
""",
|
|
roster_ids + [medal[0]],
|
|
).fetchone()
|
|
finally:
|
|
l3.close()
|
|
l2.close()
|
|
web.close()
|
|
self.assertEqual(medal[1], expected[0])
|
|
self.assertAlmostEqual(medal[2], expected[1], places=8)
|
|
|
|
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()
|