""" Empirical Challenger Test Suite for Milestone 1: Game Design Matrix & Core Contracts. Author: Challenger 2 (Empirical Challenger) Validates Protobuf schemas (party, social, portal), SQLite foreign key integrity, and database re-seeding idempotency under adversarial conditions. """ from __future__ import annotations import os import subprocess import sys import tempfile import sqlite3 import pytest from google.protobuf.descriptor_pb2 import FileDescriptorSet from server.world.game_design_matrix_service import GameDesignMatrixService PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) def test_protobuf_schemas_syntax_and_compilation(): """Compiles proto/party.proto, proto/social.proto, and proto/portal.proto via protoc.""" proto_dir = os.path.join(PROJECT_ROOT, "proto") proto_files = [ os.path.join(proto_dir, "party.proto"), os.path.join(proto_dir, "social.proto"), os.path.join(proto_dir, "portal.proto"), ] for p in proto_files: assert os.path.exists(p), f"Proto file not found: {p}" with tempfile.NamedTemporaryFile(suffix=".desc", delete=False) as f: desc_path = f.name try: cmd = [ sys.executable, "-m", "grpc_tools.protoc", f"-I{proto_dir}", f"--descriptor_set_out={desc_path}", ] + proto_files res = subprocess.run(cmd, capture_output=True, text=True) assert res.returncode == 0, f"protoc failed: {res.stderr}" with open(desc_path, "rb") as f: fds = FileDescriptorSet.FromString(f.read()) assert len(fds.file) == 3, f"Expected 3 file descriptors, got {len(fds.file)}" for f_desc in fds.file: assert f_desc.syntax == "proto3", f"{f_desc.name} is not proto3" # Validate Enum zero-based default and unique values for enum_type in f_desc.enum_type: assert len(enum_type.value) > 0, f"Enum {enum_type.name} has no values" first_val = enum_type.value[0] assert first_val.number == 0, ( f"Enum {enum_type.name} first value '{first_val.name}' has number {first_val.number} != 0" ) val_nums = [v.number for v in enum_type.value] assert len(val_nums) == len(set(val_nums)), ( f"Enum {enum_type.name} contains duplicate value numbers: {val_nums}" ) # Validate Message field numbers for msg_type in f_desc.message_type: tag_nums = [fld.number for fld in msg_type.field] assert len(tag_nums) == len(set(tag_nums)), ( f"Message {msg_type.name} contains duplicate field tags: {tag_nums}" ) for tag in tag_nums: assert tag >= 1, f"Message {msg_type.name} has invalid tag {tag} < 1" assert not (19000 <= tag <= 19999), ( f"Message {msg_type.name} uses reserved protobuf internal tag {tag}" ) finally: if os.path.exists(desc_path): os.remove(desc_path) def test_protobuf_serialization_roundtrip(): """Generates Python stubs and validates binary serialization/deserialization.""" proto_dir = os.path.join(PROJECT_ROOT, "proto") with tempfile.TemporaryDirectory() as tmpdir: cmd = [ sys.executable, "-m", "grpc_tools.protoc", f"-I{proto_dir}", f"--python_out={tmpdir}", os.path.join(proto_dir, "party.proto"), os.path.join(proto_dir, "social.proto"), os.path.join(proto_dir, "portal.proto"), ] res = subprocess.run(cmd, capture_output=True, text=True) assert res.returncode == 0, f"Python stub generation failed: {res.stderr}" sys.path.insert(0, tmpdir) try: import party_pb2 # type: ignore import social_pb2 # type: ignore import portal_pb2 # type: ignore # 1. PartyInfo serialization p_orig = party_pb2.PartyInfo( party_id="pty_test_123", leader_id="usr_leader_01", allocation_mode=party_pb2.LootAllocationMode.SHORT, short_allocation_timeout_sec=5.0, scaling_stats=party_pb2.PartyScalingStats( party_size=3, nearby_members_count=3, monster_hp_multiplier=2.0, exp_bonus_pct=60.0, item_quantity_bonus_pct=100.0, ), ) p_decoded = party_pb2.PartyInfo() p_decoded.ParseFromString(p_orig.SerializeToString()) assert p_decoded.party_id == "pty_test_123" assert p_decoded.allocation_mode == party_pb2.LootAllocationMode.SHORT assert p_decoded.scaling_stats.monster_hp_multiplier == 2.0 # 2. FriendEntry serialization f_orig = social_pb2.FriendEntry( account_id="acc_alpha", friend_account_id="acc_beta", friend_name="Chiến Lang", friend_class="char_feral_berserker", friend_level=45, presence_status=social_pb2.OnlinePresenceStatus.STATUS_ONLINE, status=social_pb2.FriendRelationStatus.ACCEPTED, friend_note="Đồng đội farm map", ) f_decoded = social_pb2.FriendEntry() f_decoded.ParseFromString(f_orig.SerializeToString()) assert f_decoded.friend_name == "Chiến Lang" assert f_decoded.friend_level == 45 assert f_decoded.presence_status == social_pb2.OnlinePresenceStatus.STATUS_ONLINE # 3. TownPortalLink serialization portal_orig = portal_pb2.TownPortalLink( portal_id="portal_link_789", owner_id="acc_alpha", wilderness_zone_id="zone_tang_kiem_nhai", wilderness_x=120.5, wilderness_y=340.25, sanctuary_zone_id="zone_player_hideout", sanctuary_x=0.0, sanctuary_y=0.0, remaining_charges=6, is_open=True, ) portal_decoded = portal_pb2.TownPortalLink() portal_decoded.ParseFromString(portal_orig.SerializeToString()) assert portal_decoded.remaining_charges == 6 assert portal_decoded.wilderness_x == 120.5 assert portal_decoded.is_open is True finally: sys.path.remove(tmpdir) def test_sqlite_foreign_key_integrity_and_rebuild_idempotency(): """Validates PRAGMA foreign_key_check and successive re-seeding idempotency.""" service = GameDesignMatrixService() try: # Successive re-seeding test for i in range(5): stats = service.seed_canonical_data(force=True) assert stats["party_scaling_matrix"] == 6 assert stats["loot_allocation_modes"] == 3 assert stats["teleport_configs"] == 4 assert stats["portal_quarantine_rules"] == 4 with service._get_connection() as conn: fk_violations = conn.cursor().execute("PRAGMA foreign_key_check").fetchall() assert len(fk_violations) == 0, f"Iteration {i} failed FK check: {fk_violations}" # Verify integrity report report = service.validate_game_design_integrity() assert report.is_valid is True assert len(report.violations) == 0 finally: service.close() def test_sqlite_table_check_constraints_adversarial(): """Adversarially attempts invalid insertions to test CHECK constraint boundaries.""" service = GameDesignMatrixService(db_path=":memory:") service.seed_canonical_data(force=True) with service._get_connection() as conn: cur = conn.cursor() # 1. party_scaling_matrix constraints for bad_size in (0, 7, -5): with pytest.raises(sqlite3.IntegrityError): cur.execute( "INSERT INTO party_scaling_matrix VALUES (?, 1.0, 0, 0, 0, 0, 15.0, 'bad')", (bad_size,), ) with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO party_scaling_matrix VALUES (1, 0.9, 0, 0, 0, 0, 15.0, 'bad hp')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO party_scaling_matrix VALUES (1, 1.0, -10, 0, 0, 0, 15.0, 'bad exp')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO party_scaling_matrix VALUES (1, 1.0, 0, 0, 0, 0, 0.0, 'bad radius')") # 2. loot_allocation_modes constraints with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO loot_allocation_modes VALUES ('X', 'X', -1.0, 0, 'desc')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO loot_allocation_modes VALUES ('X', 'X', 0.0, 2, 'desc')") # 3. teleport_configs constraints with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO teleport_configs VALUES ('X', -0.5, 10.0, 1, 1, 'desc')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO teleport_configs VALUES ('X', 3.5, -1.0, 1, 1, 'desc')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO teleport_configs VALUES ('X', 3.5, 10.0, 5, 1, 'desc')") # 4. portal_quarantine_rules constraints with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO portal_quarantine_rules VALUES ('X', 'WILD', 2, 1, 6, 'desc')") with pytest.raises(sqlite3.IntegrityError): cur.execute("INSERT INTO portal_quarantine_rules VALUES ('X', 'WILD', 1, 1, -1, 'desc')") # 5. Foreign Key cascade/restrict checks with pytest.raises(sqlite3.IntegrityError): cur.execute( "INSERT INTO zones (zone_id, act_id, name, zone_type, environment, min_level, max_level) " "VALUES ('z_phantom', 'ACT_NON_EXISTENT', 'Phantom Zone', 'WILDERNESS', 'OUTDOOR', 1, 10)" ) with pytest.raises(sqlite3.IntegrityError): cur.execute( "INSERT INTO npcs (npc_id, name, title, role, zone_id, x, y, dialogue_tree_id) " "VALUES ('npc_ghost', 'Ghost', 'Phantom', 'ROLE', 'ZONE_NON_EXISTENT', 0, 0, 'tree')" ) service.close()