From e3924f54ba4715fc20e69018c2eda9bdb5b04c5f Mon Sep 17 00:00:00 2001 From: "Teppei.F" <37261985+T3pp31@users.noreply.github.com> Date: Thu, 3 Sep 2026 07:48:03 +0900 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E3=83=87=E3=83=BC=E3=82=BF?= =?UTF-8?q?=E3=83=A2=E3=83=87=E3=83=AB=E3=82=92=E8=A8=AD=E5=AE=9A=E9=A7=86?= =?UTF-8?q?=E5=8B=95=E3=81=AE=20DataSchema=20=E3=81=AB=E6=B1=8E=E7=94=A8?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CSV/gakuseki/bt_addrs の直書きを data_schema 設定と ReceivedPayload に置き換え、フィルタ注入で blockchain→pipeline の結合を緩める。 Co-Authored-By: Claude Opus 4.6 (1M context) Co-authored-by: Cursor --- config/blockchain.json | 16 +- src/ble_blockchain/app/main.py | 44 ++++-- src/ble_blockchain/blockchain/myblock.py | 84 ++++++---- src/ble_blockchain/config/loader.py | 37 ++++- .../pipeline/delete_excess_data.py | 71 ++++++--- src/ble_blockchain/types.py | 24 +++ tests/unit/test_delete_excess_data.py | 38 +++++ tests/unit/test_filter_registered_data.py | 77 +++++++++ tests/unit/test_main_receive.py | 11 +- tests/unit/test_myblock.py | 146 +++++++++++++++++- 10 files changed, 468 insertions(+), 80 deletions(-) create mode 100644 src/ble_blockchain/types.py diff --git a/config/blockchain.json b/config/blockchain.json index 56bf563..bb7f609 100644 --- a/config/blockchain.json +++ b/config/blockchain.json @@ -4,5 +4,19 @@ "export_enabled": true, "min_verified_receives": 3, "require_content_hash_agreement": true, - "min_distinct_devices_for_aggregate": 2 + "min_distinct_devices_for_aggregate": 2, + "data_schema": { + "gakuseki_column": "gakuseki", + "bt_addr_column": "bt_addrs", + "csv_identity_rename": { + "学籍番号": "gakuseki" + }, + "output_columns": [ + "gakuseki", + "bt_addrs", + "device_name" + ], + "input_field": "gakuseki", + "output_field": "bt_addrs" + } } diff --git a/src/ble_blockchain/app/main.py b/src/ble_blockchain/app/main.py index b38f26e..8475399 100644 --- a/src/ble_blockchain/app/main.py +++ b/src/ble_blockchain/app/main.py @@ -28,6 +28,8 @@ from ble_blockchain.paths import repo_root from ble_blockchain.pipeline.delete_excess_data import delete_excess_data from ble_blockchain.pipeline.pandas_d_encode import pandas_decode, pandas_encode +from ble_blockchain.types import ReceivedPayload + RUNTIME_PROFILES_PATH = repo_root() / "config/runtime_profiles.json" @@ -70,7 +72,7 @@ def run_communication_steps( profile: dict[str, Any], settings: DeviceSettings, tanmatsu_bt_addrs: list[str], - receive_data_list: list[Any], + receive_data_list: list[ReceivedPayload], ) -> None: """Execute discoverable, send, receive, and sleep steps from a profile.""" defaults = {"sleep_seconds": 30} @@ -108,7 +110,7 @@ def run_communication_steps( def process_received_payload( raw: bytes, trusted_peer_pems: frozenset[str] -) -> list[Any]: +) -> ReceivedPayload: """Decrypt, verify, and decode one received payload.""" try: payload = unpack(raw) @@ -128,19 +130,33 @@ def process_received_payload( content_hash = payload_content_hash(plaintext) if not verified: - return [None, public_key, payload.signature, False, None, None] - - return [ - df, - public_key, - payload.signature, - verified, - payload.public_key_pem, - content_hash, - ] + return ReceivedPayload( + df=None, + public_key=public_key, + signature=payload.signature, + verified=False, + public_key_pem=None, + payload_content_hash=None, + ) + + return ReceivedPayload( + df=df, + public_key=public_key, + signature=payload.signature, + verified=verified, + public_key_pem=payload.public_key_pem, + payload_content_hash=content_hash, + ) except (ValueError, KeyError, json.JSONDecodeError) as exc: print(f"受信ペイロードの処理に失敗しました: {exc}") - return [None, None, b"", False, None, None] + return ReceivedPayload( + df=None, + public_key=None, + signature=b"", + verified=False, + public_key_pem=None, + payload_content_hash=None, + ) def run_pipeline( # pylint: disable=too-many-locals @@ -158,7 +174,7 @@ def run_pipeline( # pylint: disable=too-many-locals print(settings.tanmatsu_bt_addrs) - receive_data_list: list[Any] = [] + receive_data_list: list[ReceivedPayload] = [] run_communication_steps( profile, settings, settings.tanmatsu_bt_addrs, receive_data_list diff --git a/src/ble_blockchain/blockchain/myblock.py b/src/ble_blockchain/blockchain/myblock.py index a697ecf..ac05714 100644 --- a/src/ble_blockchain/blockchain/myblock.py +++ b/src/ble_blockchain/blockchain/myblock.py @@ -8,13 +8,16 @@ import math from collections import Counter from dataclasses import dataclass -from typing import Any +from typing import Any, Callable import pandas as pd -from ble_blockchain.config.loader import load_blockchain_config -from ble_blockchain.pipeline.delete_excess_data import filter_registered_data +from ble_blockchain.config.loader import ( + BlockchainConfig, + load_blockchain_config, +) from ble_blockchain.pipeline.pandas_d_encode import pandas_encode +from ble_blockchain.types import ReceivedPayload GENESIS_PREV_HASH = ( "747bc42088cf0b3915982af289189e8f14d3325a7d594bc2d30a7014a536cb13" @@ -59,11 +62,25 @@ def payload_content_hash(plaintext: bytes) -> str: class MyBlockChain: """Build and validate a chain of majority-adopted BLE observations.""" - def __init__(self) -> None: + def __init__( + self, + *, + filter_registered: Callable[[pd.DataFrame], pd.DataFrame] | None = None, + ) -> None: self.chain: list[dict[str, Any]] = [] self._last_threshold: int = 0 self._last_verified_count: int = 0 self._verified_entries: list[VerifiedReceive] = [] + if filter_registered is not None: + self._filter_registered = filter_registered + else: + # Lazily resolve the default so blockchain does not import + # pipeline at module load time (caller may inject a filter). + from ble_blockchain.pipeline.delete_excess_data import ( # pylint: disable=import-outside-toplevel + filter_registered_data, + ) + + self._filter_registered = filter_registered_data @property def last_majority_threshold(self) -> int: @@ -114,11 +131,12 @@ def add_new_block( self.chain.append(new_block) return new_block - def build_from_receives( # pylint: disable=too-many-locals - self, receive_data_list: list[list[Any]] + def build_from_receives( + self, receive_data_list: list[ReceivedPayload] ) -> None: - """Build blocks from verified receive tuples using majority rules.""" + """Build blocks from verified receive payloads using majority rules.""" config = load_blockchain_config() + schema = config.data_schema verified_entries = self._parse_verified_receives(receive_data_list) self._verified_entries = verified_entries @@ -135,12 +153,12 @@ def build_from_receives( # pylint: disable=too-many-locals bt_addrs_seen: set[str] = set() for entry in verified_entries: - for bt_addr in entry.df["bt_addrs"].astype(str).unique(): + for bt_addr in entry.df[schema.bt_addr_column].astype(str).unique(): bt_addrs_seen.add(bt_addr) for bt_addr in sorted(bt_addrs_seen): if config.one_block_per_bt_addr and any( - block["tran_body"]["output"]["bt_addrs"] == bt_addr + block["tran_body"]["output"][schema.output_field] == bt_addr for block in self.chain ): continue @@ -151,40 +169,39 @@ def build_from_receives( # pylint: disable=too-many-locals if adoption is None: continue - gakuseki, reporter_count, reporters, content_hash, gakuseki_votes = ( + identity, reporter_count, reporters, content_hash, identity_votes = ( adoption ) tran_meta = { "count": reporter_count, "majority_threshold": threshold, "content_hash": content_hash, - "gakuseki_votes": gakuseki_votes, + "gakuseki_votes": identity_votes, "reporters": reporters, } - inp = {"gakuseki": gakuseki} - out = {"bt_addrs": bt_addr, "count": reporter_count} + inp = {schema.input_field: identity} + out = {schema.output_field: bt_addr, "count": reporter_count} self.add_new_block(inp, out, tran_meta=tran_meta) def _parse_verified_receives( - self, receive_data_list: list[list[Any]] + self, receive_data_list: list[ReceivedPayload] ) -> list[VerifiedReceive]: verified_entries: list[VerifiedReceive] = [] for item in receive_data_list: - if len(item) < 4 or not item[3]: + if not item.verified: continue - if item[0] is None: + if item.df is None: continue - df = filter_registered_data(item[0]) + df = self._filter_registered(item.df) if df.empty: continue - if len(item) < 6: + if item.public_key_pem is None or item.payload_content_hash is None: continue - public_key_pem = str(item[4]) - declared_hash = str(item[5]) + declared_hash = str(item.payload_content_hash) encoded_hash = payload_content_hash(pandas_encode(df)) if declared_hash != encoded_hash: continue @@ -192,7 +209,7 @@ def _parse_verified_receives( verified_entries.append( VerifiedReceive( df=df, - pubkey_fingerprint=pubkey_fingerprint(public_key_pem), + pubkey_fingerprint=pubkey_fingerprint(str(item.public_key_pem)), payload_content_hash=declared_hash, ) ) @@ -204,12 +221,13 @@ def _compute_adoption_for_bt_addr( # pylint: disable=too-many-locals verified_entries: list[VerifiedReceive], bt_addr: str, threshold: int, - config: Any, + config: BlockchainConfig, ) -> tuple[str, int, list[dict[str, str]], str, dict[str, int]] | None: """Return adoption data for one bt_addr when threshold is met.""" + schema = config.data_schema reporters_for_addr: list[VerifiedReceive] = [] for entry in verified_entries: - if bt_addr in entry.df["bt_addrs"].astype(str).values: + if bt_addr in entry.df[schema.bt_addr_column].astype(str).values: reporters_for_addr.append(entry) unique_by_fingerprint: dict[str, VerifiedReceive] = {} @@ -226,21 +244,23 @@ def _compute_adoption_for_bt_addr( # pylint: disable=too-many-locals return None content_hash = next(iter(content_hashes)) - gakuseki_votes: Counter[str] = Counter() + identity_votes: Counter[str] = Counter() for entry in unique_entries: - rows = entry.df[entry.df["bt_addrs"].astype(str) == bt_addr] + rows = entry.df[ + entry.df[schema.bt_addr_column].astype(str) == bt_addr + ] if rows.empty: continue - gakuseki_votes[str(rows.iloc[0]["gakuseki"])] += 1 + identity_votes[str(rows.iloc[0][schema.gakuseki_column])] += 1 - if not gakuseki_votes: + if not identity_votes: return None - top_count = max(gakuseki_votes.values()) - winners = [g for g, c in gakuseki_votes.items() if c == top_count] + top_count = max(identity_votes.values()) + winners = [g for g, c in identity_votes.items() if c == top_count] if len(winners) != 1: return None - gakuseki = winners[0] + identity = winners[0] reporters: list[dict[str, str]] = [ { @@ -251,11 +271,11 @@ def _compute_adoption_for_bt_addr( # pylint: disable=too-many-locals ] return ( - gakuseki, + identity, reporter_count, reporters, content_hash, - dict(gakuseki_votes), + dict(identity_votes), ) def validate_chain(self) -> bool: diff --git a/src/ble_blockchain/config/loader.py b/src/ble_blockchain/config/loader.py index 219eaec..4abcd3b 100644 --- a/src/ble_blockchain/config/loader.py +++ b/src/ble_blockchain/config/loader.py @@ -2,7 +2,7 @@ import json import os -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any from ble_blockchain.paths import repo_root @@ -39,6 +39,21 @@ class PathsConfig: chain_export_dir: str +@dataclass(frozen=True) +class DataSchema: + """Column and field names for the observation data model.""" + gakuseki_column: str = "gakuseki" + bt_addr_column: str = "bt_addrs" + csv_identity_rename: dict[str, str] = field( + default_factory=lambda: {"学籍番号": "gakuseki"} + ) + output_columns: list[str] = field( + default_factory=lambda: ["gakuseki", "bt_addrs", "device_name"] + ) + input_field: str = "gakuseki" + output_field: str = "bt_addrs" + + @dataclass(frozen=True) class BlockchainConfig: """Blockchain majority and export rules.""" @@ -48,6 +63,7 @@ class BlockchainConfig: min_verified_receives: int require_content_hash_agreement: bool min_distinct_devices_for_aggregate: int + data_schema: DataSchema def load_l2cap_config() -> L2capConfig: @@ -78,6 +94,24 @@ def load_paths_config() -> PathsConfig: ) +def load_data_schema(raw: dict[str, Any]) -> DataSchema: + """Build a DataSchema from a partial data_schema config block.""" + rename = raw.get("csv_identity_rename", {"学籍番号": "gakuseki"}) + return DataSchema( + gakuseki_column=str(raw.get("gakuseki_column", "gakuseki")), + bt_addr_column=str(raw.get("bt_addr_column", "bt_addrs")), + csv_identity_rename={str(key): str(val) for key, val in rename.items()}, + output_columns=[ + str(column) + for column in raw.get( + "output_columns", ["gakuseki", "bt_addrs", "device_name"] + ) + ], + input_field=str(raw.get("input_field", "gakuseki")), + output_field=str(raw.get("output_field", "bt_addrs")), + ) + + def load_blockchain_config() -> BlockchainConfig: """Load blockchain rules from config/blockchain.json.""" data = load_json_config("blockchain.json") @@ -92,6 +126,7 @@ def load_blockchain_config() -> BlockchainConfig: min_distinct_devices_for_aggregate=int( data.get("min_distinct_devices_for_aggregate", 2) ), + data_schema=load_data_schema(data.get("data_schema", {})), ) diff --git a/src/ble_blockchain/pipeline/delete_excess_data.py b/src/ble_blockchain/pipeline/delete_excess_data.py index 1ed0ceb..eb41c8c 100644 --- a/src/ble_blockchain/pipeline/delete_excess_data.py +++ b/src/ble_blockchain/pipeline/delete_excess_data.py @@ -1,45 +1,76 @@ """Filter scan and receive DataFrames against preliminary registration data.""" +from __future__ import annotations + import pandas as pd -from ble_blockchain.config.loader import load_paths_config +from ble_blockchain.config.loader import ( + DataSchema, + load_blockchain_config, + load_paths_config, +) + +def _resolve_schema( + schema: DataSchema | None = None, +) -> DataSchema: + """Return the supplied schema or the configured default.""" + if schema is not None: + return schema + return load_blockchain_config().data_schema -def _load_preliminary_data() -> pd.DataFrame: + +def _load_preliminary_data( + schema: DataSchema | None = None, +) -> pd.DataFrame: + resolved = _resolve_schema(schema) paths = load_paths_config() preliminary_data = pd.read_csv(paths.preliminary_csv) - return preliminary_data.rename(columns={"学籍番号": "gakuseki"}) + return preliminary_data.rename(columns=resolved.csv_identity_rename) -def _load_registered_bt_addrs() -> set[str]: - preliminary_data = _load_preliminary_data() - return set(preliminary_data["bt_addrs"].astype(str)) +def _load_registered_bt_addrs( + schema: DataSchema | None = None, +) -> set[str]: + preliminary_data = _load_preliminary_data(schema) + return set(preliminary_data[_resolve_schema(schema).bt_addr_column].astype(str)) -def _merge_with_preliminary(df: pd.DataFrame) -> pd.DataFrame: - """Keep rows that match preliminary CSV on bt_addrs (and gakuseki when present).""" - preliminary_data = _load_preliminary_data() - if "gakuseki" in df.columns: +def _merge_with_preliminary( + df: pd.DataFrame, schema: DataSchema | None = None +) -> pd.DataFrame: + """Keep rows that match preliminary CSV on bt_addrs (and identity when present).""" + resolved = _resolve_schema(schema) + preliminary_data = _load_preliminary_data(resolved) + if resolved.gakuseki_column in df.columns: merged = pd.merge( preliminary_data, df, - on=["gakuseki", "bt_addrs"], + on=[resolved.gakuseki_column, resolved.bt_addr_column], how="inner", ) else: - merged = pd.merge(preliminary_data, df, on="bt_addrs", how="inner") + merged = pd.merge( + preliminary_data, + df, + on=resolved.bt_addr_column, + how="inner", + ) - columns = ["gakuseki", "bt_addrs"] - if "device_name" in merged.columns: - columns.append("device_name") - return merged[columns].drop_duplicates() + columns = list(resolved.output_columns) + present = [column for column in columns if column in merged.columns] + return merged[present].drop_duplicates() -def filter_registered_data(df: pd.DataFrame) -> pd.DataFrame: +def filter_registered_data( + df: pd.DataFrame, schema: DataSchema | None = None +) -> pd.DataFrame: """Keep registered rows for both scan results and decoded receive DataFrames.""" - return _merge_with_preliminary(df) + return _merge_with_preliminary(df, schema) -def delete_excess_data(df: pd.DataFrame) -> pd.DataFrame: +def delete_excess_data( + df: pd.DataFrame, schema: DataSchema | None = None +) -> pd.DataFrame: """Keep only rows whose bt_addrs appear in the preliminary registration CSV.""" - return _merge_with_preliminary(df) + return _merge_with_preliminary(df, schema) diff --git a/src/ble_blockchain/types.py b/src/ble_blockchain/types.py new file mode 100644 index 0000000..a276980 --- /dev/null +++ b/src/ble_blockchain/types.py @@ -0,0 +1,24 @@ +"""Shared typed data structures for the BLE blockchain pipeline.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import pandas as pd + + +@dataclass(frozen=True) +class ReceivedPayload: + """One received payload entry used for chain building. + + Replaces the former positional 6-element list contract so the + blockchain layer can read fields by name. + """ + + df: pd.DataFrame | None + public_key: Any + signature: bytes + verified: bool + public_key_pem: str | None + payload_content_hash: str | None diff --git a/tests/unit/test_delete_excess_data.py b/tests/unit/test_delete_excess_data.py index 7bd2008..78d48cf 100644 --- a/tests/unit/test_delete_excess_data.py +++ b/tests/unit/test_delete_excess_data.py @@ -2,6 +2,7 @@ import pandas as pd +from ble_blockchain.config.loader import DataSchema from ble_blockchain.pipeline.delete_excess_data import delete_excess_data @@ -35,3 +36,40 @@ def test_delete_excess_data_empty_when_no_match() -> None: assert result.empty assert "gakuseki" in result.columns assert "bt_addrs" in result.columns + + +def test_delete_excess_data_with_csv_identity_rename( + tmp_path, monkeypatch +) -> None: + """正常系: csv_identity_rename maps foreign CSV headers into schema columns.""" + # Given: CSV using Japanese header and schema rename mapping + csv_path = tmp_path / "registry.csv" + csv_path.write_text( + "学籍番号,bt_addrs,備考\n19G110001,FC:66:CF:BE:10:BF,phone\n", + encoding="utf-8", + ) + monkeypatch.setattr( + "ble_blockchain.pipeline.delete_excess_data.load_paths_config", + lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), + ) + schema = DataSchema( + gakuseki_column="gakuseki", + bt_addr_column="bt_addrs", + csv_identity_rename={"学籍番号": "gakuseki"}, + output_columns=["gakuseki", "bt_addrs", "device_name"], + input_field="gakuseki", + output_field="bt_addrs", + ) + df = pd.DataFrame( + { + "bt_addrs": ["FC:66:CF:BE:10:BF", "AA:BB:CC:DD:EE:FF"], + "device_name": ["phone", "laptop"], + } + ) + + # When: filtering with explicit schema + result = delete_excess_data(df, schema=schema) + + # Then: rename applied and only registered row kept + assert len(result) == 1 + assert result.iloc[0]["gakuseki"] == "19G110001" diff --git a/tests/unit/test_filter_registered_data.py b/tests/unit/test_filter_registered_data.py index 3a98e14..95c4fb5 100644 --- a/tests/unit/test_filter_registered_data.py +++ b/tests/unit/test_filter_registered_data.py @@ -2,6 +2,7 @@ import pandas as pd +from ble_blockchain.config.loader import DataSchema from ble_blockchain.pipeline.delete_excess_data import filter_registered_data @@ -52,3 +53,79 @@ def test_filter_registered_data_drops_wrong_gakuseki_for_registered_bt_addr() -> # Then: row removed assert result.empty + + +def test_filter_registered_data_with_alternate_schema( + tmp_path, monkeypatch +) -> None: + """正常系: alternate DataSchema drives column rename and merge keys.""" + # Given: CSV with English headers and matching schema + csv_path = tmp_path / "registry.csv" + csv_path.write_text( + "student_id,mac,note\nS001,11:22:33:44:55:66,phone\n", + encoding="utf-8", + ) + monkeypatch.setattr( + "ble_blockchain.pipeline.delete_excess_data.load_paths_config", + lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), + ) + schema = DataSchema( + gakuseki_column="student_id", + bt_addr_column="mac", + csv_identity_rename={}, + output_columns=["student_id", "mac", "device_name"], + input_field="student_id", + output_field="mac", + ) + df = pd.DataFrame( + { + "student_id": ["S001"], + "mac": ["11:22:33:44:55:66"], + "device_name": ["phone"], + } + ) + + # When: filtering with alternate schema + result = filter_registered_data(df, schema=schema) + + # Then: row kept under alternate column names + assert list(result.columns) == ["student_id", "mac", "device_name"] + assert len(result) == 1 + assert result.iloc[0]["student_id"] == "S001" + + +def test_filter_registered_data_alternate_schema_drops_unregistered( + tmp_path, monkeypatch +) -> None: + """異常系: alternate schema drops unregistered mac addresses.""" + # Given: registry without the scanned mac + csv_path = tmp_path / "registry.csv" + csv_path.write_text( + "student_id,mac,note\nS001,11:22:33:44:55:66,phone\n", + encoding="utf-8", + ) + monkeypatch.setattr( + "ble_blockchain.pipeline.delete_excess_data.load_paths_config", + lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), + ) + schema = DataSchema( + gakuseki_column="student_id", + bt_addr_column="mac", + csv_identity_rename={}, + output_columns=["student_id", "mac", "device_name"], + input_field="student_id", + output_field="mac", + ) + df = pd.DataFrame( + { + "student_id": ["S001"], + "mac": ["00:00:00:00:00:00"], + "device_name": ["unknown"], + } + ) + + # When: filtering + result = filter_registered_data(df, schema=schema) + + # Then: unregistered mac removed + assert result.empty diff --git a/tests/unit/test_main_receive.py b/tests/unit/test_main_receive.py index b7b4ed6..f132952 100644 --- a/tests/unit/test_main_receive.py +++ b/tests/unit/test_main_receive.py @@ -7,6 +7,7 @@ from ble_blockchain.cipher.aes_cipher import encrypt_payload from ble_blockchain.cipher.cipher import make_key, make_signature, public_key_to_pem from ble_blockchain.pipeline.pandas_d_encode import pandas_encode +from ble_blockchain.types import ReceivedPayload _SAMPLE_DF = pd.DataFrame( { @@ -45,8 +46,9 @@ def test_process_received_payload_rejects_untrusted_public_key() -> None: result = process_received_payload(raw, trusted) # Then: not verified and no dataframe for chain - assert result[3] is False - assert result[0] is None + assert isinstance(result, ReceivedPayload) + assert result.verified is False + assert result.df is None def test_process_received_payload_accepts_trusted_peer() -> None: @@ -58,5 +60,6 @@ def test_process_received_payload_accepts_trusted_peer() -> None: result = process_received_payload(raw, frozenset({pem})) # Then: verified with dataframe - assert result[3] is True - assert result[0] is not None + assert isinstance(result, ReceivedPayload) + assert result.verified is True + assert result.df is not None diff --git a/tests/unit/test_myblock.py b/tests/unit/test_myblock.py index 37847fa..60b2049 100644 --- a/tests/unit/test_myblock.py +++ b/tests/unit/test_myblock.py @@ -12,9 +12,10 @@ payload_content_hash, pubkey_fingerprint, ) -from ble_blockchain.config.loader import load_blockchain_config +from ble_blockchain.config.loader import DataSchema, load_blockchain_config from ble_blockchain.pipeline.delete_excess_data import filter_registered_data from ble_blockchain.pipeline.pandas_d_encode import pandas_encode +from ble_blockchain.types import ReceivedPayload _DEFAULT_PUBLIC_KEY_PEM = ( "-----BEGIN PUBLIC KEY-----\ntest\n-----END PUBLIC KEY-----" @@ -34,8 +35,8 @@ def _receive_item( *, public_key_pem: str = _DEFAULT_PUBLIC_KEY_PEM, content_hash: Optional[str] = None, -) -> list: - """Build a receive list entry for build_from_receives tests.""" +) -> ReceivedPayload: + """Build a ReceivedPayload for build_from_receives tests.""" df = pd.DataFrame( { "gakuseki": [gakuseki], @@ -45,7 +46,14 @@ def _receive_item( ) if content_hash is None: content_hash = _content_hash_for_df(df) - return [df, None, b"sig", verified, public_key_pem, content_hash] + return ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=verified, + public_key_pem=public_key_pem, + payload_content_hash=content_hash, + ) def test_build_from_receives_majority_threshold() -> None: @@ -129,7 +137,7 @@ def test_build_from_receives_below_threshold_no_block( # Given: 4 verified reporters split 2+2 across bt_addrs, threshold=3 monkeypatch.setattr( "ble_blockchain.pipeline.delete_excess_data._load_preliminary_data", - lambda: pd.DataFrame( + lambda schema=None: pd.DataFrame( { "gakuseki": ["19G110001", "19G110002"], "bt_addrs": ["FC:66:CF:BE:10:BF", "BB:BB:BB:BB:BB:BB"], @@ -318,9 +326,30 @@ def test_row_inflation_same_reporter_does_not_satisfy_majority() -> None: ) content_hash = _content_hash_for_df(df) receives = [ - [df, None, b"sig", True, "pem-a", content_hash], - [df, None, b"sig", True, "pem-a", content_hash], - [df, None, b"sig", True, "pem-a", content_hash], + ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=True, + public_key_pem="pem-a", + payload_content_hash=content_hash, + ), + ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=True, + public_key_pem="pem-a", + payload_content_hash=content_hash, + ), + ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=True, + public_key_pem="pem-a", + payload_content_hash=content_hash, + ), ] # When: building chain @@ -396,3 +425,104 @@ def test_validate_tran_meta_detects_tampered_count() -> None: errors = chain.validate_tran_meta_verbose() assert len(errors) == 1 assert "reporters length" in errors[0].reason + + +def test_filter_injection_bypasses_pipeline_csv() -> None: + """正常系: injected filter replaces pipeline CSV dependency.""" + # Given: filter that keeps rows as-is (no preliminary CSV) + def identity_filter(df: pd.DataFrame) -> pd.DataFrame: + return df + + chain = MyBlockChain(filter_registered=identity_filter) + df = pd.DataFrame( + { + "gakuseki": ["ID-X"], + "bt_addrs": ["AA:BB:CC:DD:EE:FF"], + "device_name": ["x"], + } + ) + content_hash = payload_content_hash(pandas_encode(df)) + receives = [ + ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=True, + public_key_pem=f"pem-{index}", + payload_content_hash=content_hash, + ) + for index in range(3) + ] + + # When: building with injected filter + chain.build_from_receives(receives) + + # Then: block created without preliminary CSV + assert len(chain.chain) == 1 + assert chain.chain[0]["tran_body"]["input"]["gakuseki"] == "ID-X" + assert chain.chain[0]["tran_body"]["output"]["bt_addrs"] == "AA:BB:CC:DD:EE:FF" + + +def test_data_schema_alternate_column_names( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """正常系: alternate data_schema column names drive majority adoption.""" + # Given: schema using student_id / mac instead of gakuseki / bt_addrs + alt_schema = DataSchema( + gakuseki_column="student_id", + bt_addr_column="mac", + csv_identity_rename={}, + output_columns=["student_id", "mac", "device_name"], + input_field="student_id", + output_field="mac", + ) + fake_config = load_blockchain_config() + # Rebuild with alternate schema while keeping majority rules + from ble_blockchain.config.loader import BlockchainConfig + + monkeypatch.setattr( + "ble_blockchain.blockchain.myblock.load_blockchain_config", + lambda: BlockchainConfig( + majority_ratio=fake_config.majority_ratio, + one_block_per_bt_addr=fake_config.one_block_per_bt_addr, + export_enabled=fake_config.export_enabled, + min_verified_receives=fake_config.min_verified_receives, + require_content_hash_agreement=fake_config.require_content_hash_agreement, + min_distinct_devices_for_aggregate=( + fake_config.min_distinct_devices_for_aggregate + ), + data_schema=alt_schema, + ), + ) + + def identity_filter(df: pd.DataFrame) -> pd.DataFrame: + return df + + chain = MyBlockChain(filter_registered=identity_filter) + df = pd.DataFrame( + { + "student_id": ["S001"], + "mac": ["11:22:33:44:55:66"], + "device_name": ["phone"], + } + ) + content_hash = payload_content_hash(pandas_encode(df)) + receives = [ + ReceivedPayload( + df=df, + public_key=None, + signature=b"sig", + verified=True, + public_key_pem=f"pem-{index}", + payload_content_hash=content_hash, + ) + for index in range(3) + ] + + # When: building with alternate schema + chain.build_from_receives(receives) + + # Then: block uses alternate field names + assert len(chain.chain) == 1 + assert chain.chain[0]["tran_body"]["input"]["student_id"] == "S001" + assert chain.chain[0]["tran_body"]["output"]["mac"] == "11:22:33:44:55:66" From ae871813bf1705edb20b6f702c13819ba79045d4 Mon Sep 17 00:00:00 2001 From: "Teppei.F" <37261985+T3pp31@users.noreply.github.com> Date: Thu, 3 Sep 2026 07:50:26 +0900 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20pylint=20=E6=8C=87=E6=91=98=EF=BC=88?= =?UTF-8?q?too-many-locals=20/=20=E9=87=8D=E8=A4=87=20/=20lambda=EF=BC=89?= =?UTF-8?q?=E3=82=92=E8=A7=A3=E6=B6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI の pylint 失敗を直し、テスト用 DataSchema ヘルパーを共通化した。 Co-Authored-By: Claude Opus 4.6 (1M context) Co-authored-by: Cursor --- src/ble_blockchain/blockchain/myblock.py | 2 +- tests/conftest_helpers.py | 28 +++++++++++++++++ tests/unit/test_delete_excess_data.py | 11 ++++--- tests/unit/test_filter_registered_data.py | 37 ++++++----------------- tests/unit/test_myblock.py | 25 ++++++--------- 5 files changed, 55 insertions(+), 48 deletions(-) diff --git a/src/ble_blockchain/blockchain/myblock.py b/src/ble_blockchain/blockchain/myblock.py index ac05714..968faf8 100644 --- a/src/ble_blockchain/blockchain/myblock.py +++ b/src/ble_blockchain/blockchain/myblock.py @@ -131,7 +131,7 @@ def add_new_block( self.chain.append(new_block) return new_block - def build_from_receives( + def build_from_receives( # pylint: disable=too-many-locals self, receive_data_list: list[ReceivedPayload] ) -> None: """Build blocks from verified receive payloads using majority rules.""" diff --git a/tests/conftest_helpers.py b/tests/conftest_helpers.py index dc575b8..18018ce 100644 --- a/tests/conftest_helpers.py +++ b/tests/conftest_helpers.py @@ -6,6 +6,7 @@ from conftest import valid_tran_meta from ble_blockchain.blockchain.myblock import MyBlockChain +from ble_blockchain.config.loader import DataSchema def patch_chain_export_dir( @@ -24,6 +25,33 @@ def load_paths_config() -> object: ) +def patch_preliminary_csv( + monkeypatch: pytest.MonkeyPatch, csv_path: Path +) -> None: + """Redirect preliminary CSV path used by delete_excess_data.""" + path_str = str(csv_path) + + def load_paths_config() -> object: + return type("Paths", (), {"preliminary_csv": path_str})() + + monkeypatch.setattr( + "ble_blockchain.pipeline.delete_excess_data.load_paths_config", + load_paths_config, + ) + + +def student_mac_schema() -> DataSchema: + """Return a DataSchema using student_id / mac column names.""" + return DataSchema( + gakuseki_column="student_id", + bt_addr_column="mac", + csv_identity_rename={}, + output_columns=["student_id", "mac", "device_name"], + input_field="student_id", + output_field="mac", + ) + + def make_chain_with_block( *, gakuseki: str = "19G110001", diff --git a/tests/unit/test_delete_excess_data.py b/tests/unit/test_delete_excess_data.py index 78d48cf..db4de8d 100644 --- a/tests/unit/test_delete_excess_data.py +++ b/tests/unit/test_delete_excess_data.py @@ -1,7 +1,11 @@ """Unit tests for delete_excess_data pipeline step.""" +from pathlib import Path + import pandas as pd +import pytest +from conftest_helpers import patch_preliminary_csv from ble_blockchain.config.loader import DataSchema from ble_blockchain.pipeline.delete_excess_data import delete_excess_data @@ -39,7 +43,7 @@ def test_delete_excess_data_empty_when_no_match() -> None: def test_delete_excess_data_with_csv_identity_rename( - tmp_path, monkeypatch + tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """正常系: csv_identity_rename maps foreign CSV headers into schema columns.""" # Given: CSV using Japanese header and schema rename mapping @@ -48,10 +52,7 @@ def test_delete_excess_data_with_csv_identity_rename( "学籍番号,bt_addrs,備考\n19G110001,FC:66:CF:BE:10:BF,phone\n", encoding="utf-8", ) - monkeypatch.setattr( - "ble_blockchain.pipeline.delete_excess_data.load_paths_config", - lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), - ) + patch_preliminary_csv(monkeypatch, csv_path) schema = DataSchema( gakuseki_column="gakuseki", bt_addr_column="bt_addrs", diff --git a/tests/unit/test_filter_registered_data.py b/tests/unit/test_filter_registered_data.py index 95c4fb5..2bc59dc 100644 --- a/tests/unit/test_filter_registered_data.py +++ b/tests/unit/test_filter_registered_data.py @@ -1,8 +1,11 @@ """Unit tests for filter_registered_data pipeline step.""" +from pathlib import Path + import pandas as pd +import pytest -from ble_blockchain.config.loader import DataSchema +from conftest_helpers import patch_preliminary_csv, student_mac_schema from ble_blockchain.pipeline.delete_excess_data import filter_registered_data @@ -56,7 +59,7 @@ def test_filter_registered_data_drops_wrong_gakuseki_for_registered_bt_addr() -> def test_filter_registered_data_with_alternate_schema( - tmp_path, monkeypatch + tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """正常系: alternate DataSchema drives column rename and merge keys.""" # Given: CSV with English headers and matching schema @@ -65,18 +68,8 @@ def test_filter_registered_data_with_alternate_schema( "student_id,mac,note\nS001,11:22:33:44:55:66,phone\n", encoding="utf-8", ) - monkeypatch.setattr( - "ble_blockchain.pipeline.delete_excess_data.load_paths_config", - lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), - ) - schema = DataSchema( - gakuseki_column="student_id", - bt_addr_column="mac", - csv_identity_rename={}, - output_columns=["student_id", "mac", "device_name"], - input_field="student_id", - output_field="mac", - ) + patch_preliminary_csv(monkeypatch, csv_path) + schema = student_mac_schema() df = pd.DataFrame( { "student_id": ["S001"], @@ -95,7 +88,7 @@ def test_filter_registered_data_with_alternate_schema( def test_filter_registered_data_alternate_schema_drops_unregistered( - tmp_path, monkeypatch + tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """異常系: alternate schema drops unregistered mac addresses.""" # Given: registry without the scanned mac @@ -104,18 +97,8 @@ def test_filter_registered_data_alternate_schema_drops_unregistered( "student_id,mac,note\nS001,11:22:33:44:55:66,phone\n", encoding="utf-8", ) - monkeypatch.setattr( - "ble_blockchain.pipeline.delete_excess_data.load_paths_config", - lambda: type("P", (), {"preliminary_csv": str(csv_path)})(), - ) - schema = DataSchema( - gakuseki_column="student_id", - bt_addr_column="mac", - csv_identity_rename={}, - output_columns=["student_id", "mac", "device_name"], - input_field="student_id", - output_field="mac", - ) + patch_preliminary_csv(monkeypatch, csv_path) + schema = student_mac_schema() df = pd.DataFrame( { "student_id": ["S001"], diff --git a/tests/unit/test_myblock.py b/tests/unit/test_myblock.py index 60b2049..8fb0313 100644 --- a/tests/unit/test_myblock.py +++ b/tests/unit/test_myblock.py @@ -6,13 +6,14 @@ import pytest from conftest import valid_tran_meta +from conftest_helpers import student_mac_schema from ble_blockchain.blockchain.myblock import ( MyBlockChain, compute_majority_threshold, payload_content_hash, pubkey_fingerprint, ) -from ble_blockchain.config.loader import DataSchema, load_blockchain_config +from ble_blockchain.config.loader import BlockchainConfig, load_blockchain_config from ble_blockchain.pipeline.delete_excess_data import filter_registered_data from ble_blockchain.pipeline.pandas_d_encode import pandas_encode from ble_blockchain.types import ReceivedPayload @@ -468,21 +469,11 @@ def test_data_schema_alternate_column_names( ) -> None: """正常系: alternate data_schema column names drive majority adoption.""" # Given: schema using student_id / mac instead of gakuseki / bt_addrs - alt_schema = DataSchema( - gakuseki_column="student_id", - bt_addr_column="mac", - csv_identity_rename={}, - output_columns=["student_id", "mac", "device_name"], - input_field="student_id", - output_field="mac", - ) + alt_schema = student_mac_schema() fake_config = load_blockchain_config() - # Rebuild with alternate schema while keeping majority rules - from ble_blockchain.config.loader import BlockchainConfig - monkeypatch.setattr( - "ble_blockchain.blockchain.myblock.load_blockchain_config", - lambda: BlockchainConfig( + def load_alt_config() -> BlockchainConfig: + return BlockchainConfig( majority_ratio=fake_config.majority_ratio, one_block_per_bt_addr=fake_config.one_block_per_bt_addr, export_enabled=fake_config.export_enabled, @@ -492,7 +483,11 @@ def test_data_schema_alternate_column_names( fake_config.min_distinct_devices_for_aggregate ), data_schema=alt_schema, - ), + ) + + monkeypatch.setattr( + "ble_blockchain.blockchain.myblock.load_blockchain_config", + load_alt_config, ) def identity_filter(df: pd.DataFrame) -> pd.DataFrame: