From 055a0a22528c1c7e18e2a144da6864d971462ccf Mon Sep 17 00:00:00 2001 From: faildeny Date: Tue, 27 Oct 2026 16:44:32 +0100 Subject: [PATCH 1/5] Add tests for all models and CI --- .github/workflows/python-ci.yml | 147 -------------------- .github/workflows/tests.yml | 55 ++++++++ pytest.ini | 5 + requirements-dev.txt | 2 + tests/conftest.py | 16 +++ tests/fl_runner.py | 232 +++++++++++++++++++++++++++++++ tests/synthetic_data.py | 212 ++++++++++++++++++++++++++++ tests/test_federated_training.py | 136 ++++++++++++++++++ tests/test_synthetic_data.py | 130 +++++++++++++++++ 9 files changed, 788 insertions(+), 147 deletions(-) delete mode 100644 .github/workflows/python-ci.yml create mode 100644 .github/workflows/tests.yml create mode 100644 pytest.ini create mode 100644 requirements-dev.txt create mode 100644 tests/conftest.py create mode 100644 tests/fl_runner.py create mode 100644 tests/synthetic_data.py create mode 100644 tests/test_federated_training.py create mode 100644 tests/test_synthetic_data.py diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml deleted file mode 100644 index 12460f1..0000000 --- a/.github/workflows/python-ci.yml +++ /dev/null @@ -1,147 +0,0 @@ -# This GitHub Action workflow will - -# - Format code using Black -# - Build project with all dependencies -# - Check code with flake8 linter -# - Run tests with pytest - - -# ubuntu-latest has 7GB RAM and 14GB SSD. macos-latest has 14GB RAM and 14GB SSD. Using macos assuming more RAM -> faster tests -> less timeouts -# Nonetheless, our tests seem to need >10 GB RAM and >16,7 GB SSD Space. Test refactoring needed. -# https://docs.github.com/en/actions/using-github-hosted-runners/about-github-hosted-runners#supported-runners-and-hardware-resources - -name: Build and test - -on: - push: - branches: [ main ] - pull_request: - branches: [ main ] - -jobs: - formatting: - outputs: - new_sha: ${{ steps.sha.outputs.SHA }} - runs-on: macos-latest - - steps: - - name: Checkout code - uses: actions/checkout@v3 - with: - ref: ${{ github.event.pull_request.head.ref }} # ${{ github.event.pull_request.head.sha }} - - - name: Setup Python 3.10 - uses: actions/setup-python@v2 - with: - python-version: "3.10" - cache: 'pip' - - - name: Setup isort and Black - run: | - python -m pip install --upgrade pip - pip install isort black - - - name: Run isort - run: | - isort --profile black . - - - name: Run Black formatter - run: | - black . - - # - name: Commit changes - # uses: EndBug/add-and-commit@v8 - # with: - # message: 'Fix styling' - # add: '*.py' - - # - name: Get commit hash - # id: sha - # run: | - # sha_new=$(git rev-parse HEAD) - # echo $sha_new - # echo "::set-output name=SHA::$sha_new" - - - build: - if: always() - needs: formatting - runs-on: macos-latest - - steps: - - name: Checkout to latest changes - uses: actions/checkout@v3 - with: - ref: ${{ needs.formatting.outputs.new_sha }} - fetch-depth: 0 - - - name: Set up Python 3.10 - uses: actions/setup-python@v2 - with: - python-version: "3.10" - cache: 'pip' - - - name: Install dependencies - run: | - python -m pip install --upgrade pip - pip install flake8 pytest pytest-cov pytest-github-actions-annotate-failures - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - pip install -e . - - lint: - needs: build - runs-on: macos-latest - - steps: - - name: Checkout to latest changes - uses: actions/checkout@v3 - with: - ref: ${{ needs.formatting.outputs.new_sha }} - fetch-depth: 0 - - - name: Set up Python 3.10 - uses: actions/setup-python@v2 - with: - python-version: "3.10" - cache: 'pip' - - - name: Install cached dependencies - run: | - python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - pip install flake8 - pip install -e . - - - name: Lint with flake8 - run: | - # stop the build if there are Python syntax errors or undefined names - flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics --max-line-length=88 --extend-ignore=E203,E501 - # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=88 --extend-ignore=E203,E501 --statistics - - test: - needs: build - runs-on: macos-latest #windows-latest - - steps: - - name: Checkout to latest changes - uses: actions/checkout@v3 - with: - ref: ${{ needs.formatting.outputs.new_sha }} - fetch-depth: 0 - - - name: Set up Python 3.10 - uses: actions/setup-python@v2 - with: - python-version: "3.10" - cache: 'pip' - - - name: Install cached dependencies - run: | - python -m pip install --upgrade pip - if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - pip install pytest pytest-cov pytest-github-actions-annotate-failures - pip install -e . - - name: Test with pytest - run: | - pytest tests -W ignore::DeprecationWarning --verbose --failed-first --log-cli-level=INFO #--cov=. --cov-report html #--capture=tee-sys \ No newline at end of file diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..1e52602 --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,55 @@ +name: Build and test + +on: + push: + branches: [main] + pull_request: + branches: [main, cmd_line] + workflow_dispatch: + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + build-and-test: + runs-on: ubuntu-latest + timeout-minutes: 60 + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up Python 3.10 + uses: actions/setup-python@v5 + with: + python-version: "3.10" + cache: pip + cache-dependency-path: | + requirements.txt + requirements-dev.txt + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + # CPU-only torch wheel: avoids the multi-GB CUDA download on the runner + pip install -r requirements.txt -r requirements-dev.txt \ + --extra-index-url https://download.pytorch.org/whl/cpu + + - name: Check that the framework imports + run: | + python -c "import flcore.utils" + python server_cmd.py --help > /dev/null + python client_cmd.py --help > /dev/null + + - name: Run tests + run: pytest tests --verbose --durations=10 + + - name: Upload federation logs on failure + if: failure() + uses: actions/upload-artifact@v4 + with: + name: pytest-tmp + path: /tmp/pytest-of-runner/ + if-no-files-found: ignore + retention-days: 7 diff --git a/pytest.ini b/pytest.ini new file mode 100644 index 0000000..922dcf7 --- /dev/null +++ b/pytest.ini @@ -0,0 +1,5 @@ +[pytest] +testpaths = tests +addopts = -ra +filterwarnings = + ignore::DeprecationWarning diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..7d0a115 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +pytest>=8 +pytest-xdist>=3 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..cf19c0e --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,16 @@ +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +from synthetic_data import DEFAULT_N_SAMPLES, write_dataset # noqa: E402 + + +@pytest.fixture(scope="session") +def synthetic_data_dir(tmp_path_factory): + """Directory with ``data.parquet`` + ``metadata.json`` of 100 synthetic rows.""" + path = tmp_path_factory.mktemp("synthetic_dt4h") + write_dataset(path, n_samples=DEFAULT_N_SAMPLES, seed=0) + return path diff --git a/tests/fl_runner.py b/tests/fl_runner.py new file mode 100644 index 0000000..56a6a60 --- /dev/null +++ b/tests/fl_runner.py @@ -0,0 +1,232 @@ +"""Run a complete local federation (1 server + N clients) as subprocesses. + +Mirrors what ``run.py`` does, but takes its parameters as arguments instead +of a YAML file so tests can use isolated sandboxes and free ports. +""" + +import socket +import subprocess +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path + +import yaml + +ROOT = Path(__file__).resolve().parent.parent + +TRAIN_LABELS = [ + "patient_demographics_gender", + "encounters_encounterClass", + "patient_demographics_age", + "encounters_lengthOfStay", + "vital_signs_weight_value_p6mo_min", + "med_admins_history_diuretics_loop_any", + "patient_flags_smoker", +] + + +@dataclass +class FederationResult: + ok: bool + reason: str + sandbox: Path + server_log: str = "" + client_logs: list = field(default_factory=list) + history: dict = field(default_factory=dict) + # True for failures of the local networking (port collision), not of the model + flaky: bool = False + + def describe(self, tail=40): + def last(text): + return "\n".join(text.strip().splitlines()[-tail:]) + + parts = [self.reason, "--- server (tail) ---", last(self.server_log)] + for i, log in enumerate(self.client_logs): + parts += [f"--- client {i} (tail) ---", last(log)] + return "\n".join(parts) + + +def free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _opt(name, value): + if value is None: + return [] + if isinstance(value, (list, tuple)): + return [f"--{name}", *[str(v) for v in value]] if value else [] + return [f"--{name}", str(value)] + + +def _wait_for_port(port, server, timeout): + """True once the server accepts connections on 127.0.0.1 (the clients' route).""" + start = time.time() + while time.time() - start < timeout: + if server.poll() is not None: + return False + try: + with socket.create_connection(("127.0.0.1", port), timeout=1): + return True + except OSError: + time.sleep(0.25) + return False + + +def _read(path): + return path.read_text(errors="replace") if path.exists() else "" + + +def _kill(processes): + for p in processes: + if p.poll() is None: + p.kill() + for p in processes: + p.wait() + + +def _attempt( + *, + model, + task, + data_path, + sandbox, + target_labels, + train_labels=TRAIN_LABELS, + num_clients=2, + num_rounds=2, + extra_args=(), + client_extra_args=(), + timeout=300, + seed=42, +): + """Train ``model`` for ``num_rounds`` with ``num_clients`` clients. + + ``extra_args`` go to both server and clients, ``client_extra_args`` only to + the clients. Never raises on a failed training: returns a + :class:`FederationResult` so the caller decides what to assert. + """ + sandbox = Path(sandbox) + sandbox.mkdir(parents=True, exist_ok=True) + # The deployed /sandbox volume ships with these directories, and some + # servers (e.g. the linear-model aggregator) do not create them on their own. + for sub in ("checkpoints", "models"): + (sandbox / sub).mkdir(exist_ok=True) + port = free_port() + + common = [ + *_opt("model", model), + *_opt("task", task), + *_opt("num_rounds", num_rounds), + *_opt("num_clients", num_clients), + *_opt("seed", seed), + *_opt("local_port", port), + *_opt("production_mode", False), + *_opt("sandbox_path", sandbox), + # run.py always forwards these two (random forest requires --balanced) + *_opt("balanced", True), + *_opt("levelOfDetail", "DecisionTree"), + *extra_args, + ] + server_cmd = [sys.executable, "server_cmd.py", *common] + client_cmd = [ + sys.executable, + "client_cmd.py", + *common, + *_opt("dataset", "dt4h_format"), + *_opt("data_path", data_path), + *_opt("data_id", data_path), + *_opt("normalization_method", "IQR"), + *_opt("train_labels", train_labels), + *_opt("target_labels", target_labels), + *_opt("train_size", 0.7), + *client_extra_args, + ] + + logs = sandbox / "logs" + logs.mkdir(exist_ok=True) + server_log = logs / "server.log" + client_logs = [logs / f"client_{i}.log" for i in range(num_clients)] + + processes = [] + + def result(ok, reason, flaky=False): + _kill(processes) + history_file = sandbox / "history.yaml" + history = {} + if history_file.exists(): + history = yaml.safe_load(history_file.read_text()) or {} + return FederationResult( + ok=ok, + reason=reason, + sandbox=sandbox, + server_log=_read(server_log), + client_logs=[_read(p) for p in client_logs], + history=history, + flaky=flaky, + ) + + with open(server_log, "w") as sf: + server = subprocess.Popen( + server_cmd, cwd=ROOT, stdout=sf, stderr=subprocess.STDOUT + ) + processes.append(server) + if not _wait_for_port(port, server, timeout=90): + # Still running but unreachable: a socket problem worth one retry. + # An exited server crashed on its own, which is a real failure. + return result( + False, "server did not start listening", flaky=server.poll() is None + ) + + client_files = [open(p, "w") for p in client_logs] + try: + clients = [] + for f in client_files: + clients.append( + subprocess.Popen( + client_cmd, cwd=ROOT, stdout=f, stderr=subprocess.STDOUT + ) + ) + processes.extend(clients) + + start = time.time() + clients_done_at = None + while server.poll() is None: + if time.time() - start > timeout: + return result(False, f"timed out after {timeout}s") + if clients_done_at is None and all(c.poll() == 0 for c in clients): + clients_done_at = time.time() + if clients_done_at and time.time() - clients_done_at > 30: + # Clients finished cleanly but the server never terminated. + return result(False, "server did not exit after training") + time.sleep(0.5) + # Clients exit right after a healthy run; if the server crashed they + # may linger retrying, so only wait for them briefly in that case. + grace = 60 if server.returncode == 0 else 5 + for c in clients: + try: + c.wait(timeout=grace) + except subprocess.TimeoutExpired: + c.kill() + c.wait() + finally: + for f in client_files: + f.close() + + codes = {"server": server.returncode} + codes.update({f"client {i}": c.returncode for i, c in enumerate(clients)}) + failed = {k: v for k, v in codes.items() if v != 0} + if failed: + return result(False, f"non-zero exit codes: {failed}") + return result(True, "ok") + + +def run_federation(**kwargs): + """:func:`_attempt` with one retry when the local sockets misbehaved.""" + res = _attempt(**kwargs) + unreachable = "failed to connect to all addresses" in "".join(res.client_logs) + if not res.ok and (res.flaky or unreachable): + res = _attempt(**kwargs) + return res diff --git a/tests/synthetic_data.py b/tests/synthetic_data.py new file mode 100644 index 0000000..b719014 --- /dev/null +++ b/tests/synthetic_data.py @@ -0,0 +1,212 @@ +"""Synthetic dataset generator that mirrors the DT4H dataset layout. + +The real datasets cannot be shared, so the tests build their own from the +structure of the sample files (``synthetic_dt4h_dataset`` and +``synthetic_survival``): + +* a table with the same 12 columns and dtypes (string / int / float / bool), + plus the 7 ``cause_of_death_isAllCause_*`` horizon columns used by the + survival models; +* a ``metadata.json`` with the same ``entries[0]`` layout (features, outcomes, + datasetStats) whose statistics are computed from the generated table. + +The generator is deterministic for a given seed, and guarantees that every +class / event appears often enough for each outcome to be trainable on a +70/30 split, whatever the sample size is (>= ~60 rows). +""" + +import json +from pathlib import Path + +import numpy as np +import pandas as pd + +DEFAULT_N_SAMPLES = 100 + +GENDERS = ["female", "male"] +ENCOUNTER_CLASSES = ["inpatient", "outpatient", "emergency"] +ADMISSION_YEARS = ["2019", "2020", "2021", "2022", "2023"] +DISPOSITIONS = ["home", "rehab", "nursing_facility", "death"] + +FEATURES = [ + ("patient_demographics_gender", "NOMINAL"), + ("encounters_encounterClass", "NOMINAL"), + ("encounters_admissionYear", "NOMINAL"), + ("patient_demographics_age", "NUMERIC"), + ("encounters_lengthOfStay", "NUMERIC"), + ("vital_signs_weight_value_p6mo_min", "NUMERIC"), + ("med_admins_history_diuretics_loop_any", "BOOLEAN"), + ("patient_flags_smoker", "BOOLEAN"), +] +OUTCOMES = [ + ("outcome_in_hospital_mortality", "BOOLEAN"), + ("outcome_readmission_30d", "BOOLEAN"), + ("outcome_discharge_disposition", "NOMINAL"), + ("outcome_severity_score", "NUMERIC"), +] + +# Accumulative survival horizons, in the same order and naming as the sample +# survival file (suffix -> days). +SURVIVAL_PREFIX = "cause_of_death_isAllCause_" +SURVIVAL_HORIZONS = { + "7d": 7, + "1mo": 30, + "3mo": 90, + "6mo": 180, + "1a": 365, + "3a": 365 * 3, + "5a": 365 * 5, +} + +FEATURE_COLUMNS = [name for name, _ in FEATURES] +OUTCOME_COLUMNS = [name for name, _ in OUTCOMES] +SURVIVAL_COLUMNS = [SURVIVAL_PREFIX + suffix for suffix in SURVIVAL_HORIZONS] +ALL_COLUMNS = FEATURE_COLUMNS + OUTCOME_COLUMNS + SURVIVAL_COLUMNS + + +def _z(x): + x = np.asarray(x, dtype=float) + return (x - x.mean()) / (x.std() or 1.0) + + +def generate_dataset(n_samples=DEFAULT_N_SAMPLES, seed=0): + """Return a DataFrame with ``n_samples`` rows and the DT4H column layout.""" + rng = np.random.default_rng(seed) + n = n_samples + + gender = rng.choice(GENDERS, n, p=[0.36, 0.64]) + encounter = rng.choice(ENCOUNTER_CLASSES, n, p=[0.42, 0.35, 0.23]) + year = rng.choice([int(y) for y in ADMISSION_YEARS], n) + age = np.clip(rng.normal(58, 16, n), 18, 97).round().astype(int) + stay = np.clip(rng.lognormal(1.3, 0.7, n), 1, 45).round().astype(int) + weight = np.clip(rng.normal(76.8, 14, n), 42.5, 135.0).round(1) + diuretics = rng.random(n) < 0.12 + smoker = rng.random(n) < 0.23 + + # One latent severity drives every outcome so the models have signal to learn. + risk = ( + 0.9 * _z(age) + + 0.7 * _z(stay) + + 0.6 * diuretics + + 0.5 * (encounter == "emergency") + + 0.3 * smoker + + rng.normal(0, 0.6, n) + ) + + # Quantile thresholds (instead of random draws) pin the class balance. + mortality = risk > np.quantile(risk, 0.80) + readmission = (risk + rng.normal(0, 0.8, n)) > np.quantile(risk, 0.82) + + # Discharge: death == mortality, remaining patients split by risk rank. + disposition = np.full(n, "home", dtype=object) + survivors = np.flatnonzero(~mortality) + ranked = survivors[np.argsort(risk[survivors])] + n_home = int(round(len(ranked) * 0.65)) + n_rehab = int(round(len(ranked) * 0.20)) + disposition[ranked[n_home:n_home + n_rehab]] = "rehab" + disposition[ranked[n_home + n_rehab:]] = "nursing_facility" + disposition[mortality] = "death" + + severity = np.clip(4.8 + 1.8 * _z(risk) + rng.normal(0, 0.8, n), 0.0, 10.0).round(1) + + # Survival: time to death, shorter with higher risk; ~40% die within 5 years. + time_to_death = rng.exponential(1.0, n) * np.exp(-0.8 * _z(risk)) * 365 * 9 + horizons = { + SURVIVAL_PREFIX + suffix: time_to_death <= days + for suffix, days in SURVIVAL_HORIZONS.items() + } + + data = { + "patient_demographics_gender": gender, + "encounters_encounterClass": encounter, + "encounters_admissionYear": year, + "patient_demographics_age": age, + "encounters_lengthOfStay": stay, + "vital_signs_weight_value_p6mo_min": weight, + "med_admins_history_diuretics_loop_any": diuretics, + "patient_flags_smoker": smoker, + "outcome_in_hospital_mortality": mortality, + "outcome_readmission_30d": readmission, + "outcome_discharge_disposition": disposition, + "outcome_severity_score": severity, + **horizons, + } + return pd.DataFrame(data)[ALL_COLUMNS] + + +def _numeric_stats(col): + col = col.dropna().astype(float) + q1, q2, q3 = np.quantile(col, [0.25, 0.5, 0.75]) + counts, edges = np.histogram(col, bins=5) + return { + "numOfNotNull": int(col.size), + "min": float(col.min()), + "max": float(col.max()), + "avg": float(round(col.mean(), 3)), + "q1": float(q1), + "q2": float(q2), + "q3": float(q3), + "histogram": [ + {"bin": float(round(edges[i + 1], 3)), "count": int(counts[i])} + for i in range(len(counts)) + ], + } + + +def _nominal_stats(col, value_set): + as_str = col.dropna().astype(str) + return { + "numOfNotNull": int(as_str.size), + "valueSet": list(value_set), + "cardinalityPerItem": {v: int((as_str == v).sum()) for v in value_set}, + } + + +def _boolean_stats(col): + col = col.dropna() + return {"numOfNotNull": int(col.size), "numOfTrue": int(col.astype(bool).sum())} + + +def build_metadata(df): + """Build a ``metadata.json`` structure (``entries[0]`` layout) for ``df``.""" + value_sets = { + "patient_demographics_gender": GENDERS, + "encounters_encounterClass": ENCOUNTER_CLASSES, + "encounters_admissionYear": ADMISSION_YEARS, + "outcome_discharge_disposition": DISPOSITIONS, + } + + def stats_for(name, dtype): + if dtype == "NUMERIC": + return _numeric_stats(df[name]) + if dtype == "NOMINAL": + return _nominal_stats(df[name], value_sets[name]) + return _boolean_stats(df[name]) + + return { + "entries": [ + { + "features": [{"name": n, "dataType": t} for n, t in FEATURES], + "outcomes": [{"name": n, "dataType": t} for n, t in OUTCOMES], + "datasetStats": { + "featureStats": {n: stats_for(n, t) for n, t in FEATURES}, + "outcomeStats": {n: stats_for(n, t) for n, t in OUTCOMES}, + }, + } + ] + } + + +def write_dataset(directory, n_samples=DEFAULT_N_SAMPLES, seed=0): + """Write ``data.parquet`` + ``metadata.json`` into ``directory``. + + The directory has the layout flcore expects for ``--data_path`` (a single + parquet file next to ``metadata.json``). Returns the generated DataFrame. + """ + directory = Path(directory) + directory.mkdir(parents=True, exist_ok=True) + df = generate_dataset(n_samples, seed) + df.to_parquet(directory / "data.parquet", index=False) + with open(directory / "metadata.json", "w") as f: + json.dump(build_metadata(df), f, indent=2) + return df diff --git a/tests/test_federated_training.py b/tests/test_federated_training.py new file mode 100644 index 0000000..dcd02b2 --- /dev/null +++ b/tests/test_federated_training.py @@ -0,0 +1,136 @@ +"""Every model, in every training scenario that applies to it. + +Each case launches a real federation (1 server + 2 clients, 2 rounds) over +the synthetic dataset and checks that every process exits cleanly and that the +server completed both rounds with finite losses. +""" + +import math + +import pytest +from fl_runner import run_federation + +NUM_CLIENTS = 2 +NUM_ROUNDS = 2 + +BINARY = ["outcome_in_hospital_mortality"] # BOOLEAN -> binary classification +MULTICLASS = ["outcome_discharge_disposition"] # NOMINAL (4) -> multiclass +REGRESSION = ["outcome_severity_score"] # NUMERIC -> regression +SURVIVAL_PATTERN = "cause_of_death_isAllCause_" # 7d, 1mo, ... 5a horizon columns + +TARGETS = { + "binary": ("classification", BINARY), + "multiclass": ("classification", MULTICLASS), + "regression": ("regression", REGRESSION), + "survival": ("survival", []), +} + + +def known_bug(reason): + """Mark a scenario that currently fails because of a bug in flcore. + + Strict, so the suite goes red as soon as the bug is fixed and this mark + needs to be removed. Run with ``--runxfail`` to see the actual failures. + """ + return pytest.mark.xfail(reason=reason, strict=True) + + +def case(model, scenario, client_extra=(), id_suffix="", bug=None): + marks = [known_bug(bug)] if bug else [] + return pytest.param( + model, + scenario, + list(client_extra), + id=f"{model}{id_suffix}-{scenario}", + marks=marks, + ) + + +SGD_MULTICLASS_BUG = ( + "linear_models.utils.set_initial_params sets a scalar intercept_ for " + "lsvc/svm, but SGDClassifier needs one intercept per class (multiclass)" +) +SVR_KERNEL_BUG = "SVR with a non-linear kernel has no fit_intercept / coef_" +XGB_BUG = "xgb server needs iteration_indptr (XGBoost >= 2.0, 1.7.5 is pinned)" +SURVIVAL_TIME_BUG = "survival client fit() uses `time` without importing it" + +CASES = [] + +# Linear classifiers: binary + multiclass +for model in ["logistic_regression", "logistic_regression_elasticnet", "lsvc", "svm"]: + sgd = model in ("lsvc", "svm") + CASES.append(case(model, "binary")) + CASES.append(case(model, "multiclass", bug=SGD_MULTICLASS_BUG if sgd else None)) + +# Linear regressors +for model in ["linear_regression", "lasso_regression", "ridge_regression"]: + CASES.append(case(model, "regression")) +CASES.append( + case( + "linear_regression_elasticnet", + "regression", + bug="server GetModelServerStrategy rejects it as 'Unknown model'", + ) +) +for kernel in ["linear", "rbf", "poly", "sigmoid"]: + CASES.append( + case( + "svr", + "regression", + client_extra=["--kernel", kernel], + id_suffix=f"[{kernel}]", + bug=None if kernel == "linear" else SVR_KERNEL_BUG, + ) + ) + +# Tree ensembles and neural network +CASES += [ + case("random_forest", "binary"), + case( + "random_forest", + "multiclass", + bug="random_forest client.evaluate crashes on multiclass probabilities", + ), + case("random_forest", "regression"), +] +wrf_bug = "weighted_random_forest server reads config['weighted_random_forest'], never set by the CLI" +CASES += [ + case("weighted_random_forest", "binary", bug=wrf_bug), + case("weighted_random_forest", "multiclass", bug=wrf_bug), +] +CASES += [case("xgb", s, bug=XGB_BUG) for s in ("binary", "multiclass", "regression")] +CASES += [case("nn", s) for s in ("binary", "multiclass", "regression")] + +# Survival +CASES += [ + case("cox", "survival", bug=SURVIVAL_TIME_BUG), + case("rsf", "survival"), + case("gbs", "survival", bug=SURVIVAL_TIME_BUG), +] + + +@pytest.mark.parametrize("model,scenario,client_extra", CASES) +def test_federated_training( + model, scenario, client_extra, synthetic_data_dir, tmp_path +): + task, target_labels = TARGETS[scenario] + if scenario == "survival": + client_extra = [*client_extra, "--accumulative_pattern_col", SURVIVAL_PATTERN] + result = run_federation( + model=model, + task=task, + data_path=synthetic_data_dir, + sandbox=tmp_path / "sandbox", + target_labels=target_labels, + num_clients=NUM_CLIENTS, + num_rounds=NUM_ROUNDS, + client_extra_args=client_extra, + ) + assert result.ok, result.describe() + + losses = result.history.get("losses_distributed", []) + assert len(losses) == NUM_ROUNDS, ( + f"expected {NUM_ROUNDS} evaluated rounds, got {len(losses)}\n" + + result.describe() + ) + assert all(math.isfinite(x) for x in losses), f"non-finite losses: {losses}" diff --git a/tests/test_synthetic_data.py b/tests/test_synthetic_data.py new file mode 100644 index 0000000..4397a6b --- /dev/null +++ b/tests/test_synthetic_data.py @@ -0,0 +1,130 @@ +"""The synthetic generator must reproduce the layout of the DT4H sample files.""" + +import json + +import numpy as np +import pandas as pd +import pytest +from synthetic_data import ( + ALL_COLUMNS, + DEFAULT_N_SAMPLES as N_SAMPLES, + DISPOSITIONS, + SURVIVAL_COLUMNS, + generate_dataset, + write_dataset, +) + +# Dtypes of synthetic_dt4h_dataset / synthetic_survival (strings are "object" +# or "str" depending on the pandas version, so they are compared by kind). +EXPECTED_KINDS = { + "patient_demographics_gender": "O", + "encounters_encounterClass": "O", + "encounters_admissionYear": "i", + "patient_demographics_age": "i", + "encounters_lengthOfStay": "i", + "vital_signs_weight_value_p6mo_min": "f", + "med_admins_history_diuretics_loop_any": "b", + "patient_flags_smoker": "b", + "outcome_in_hospital_mortality": "b", + "outcome_readmission_30d": "b", + "outcome_discharge_disposition": "O", + "outcome_severity_score": "f", + **{c: "b" for c in SURVIVAL_COLUMNS}, +} + + +@pytest.fixture(scope="module") +def df(): + return generate_dataset(N_SAMPLES, seed=0) + + +def test_has_100_rows_and_sample_file_columns(df): + assert len(df) == 100 + assert list(df.columns) == ALL_COLUMNS + assert len(ALL_COLUMNS) == 19 + + +def test_dtypes_match_sample_files(df): + for col, kind in EXPECTED_KINDS.items(): + assert df[col].dtype.kind == kind, f"{col}: {df[col].dtype}" + + +def test_no_missing_values(df): + assert not df.isna().any().any() + + +def test_value_ranges_match_metadata_sample(df): + assert df["patient_demographics_age"].between(18, 97).all() + assert df["encounters_lengthOfStay"].between(1, 45).all() + assert df["vital_signs_weight_value_p6mo_min"].between(42.5, 135.0).all() + assert df["outcome_severity_score"].between(0.0, 10.0).all() + assert set(df["outcome_discharge_disposition"]) == set(DISPOSITIONS) + + +def test_every_outcome_is_trainable_on_a_70_30_split(df): + # Each class needs enough members to land in both train and test splits. + assert df["outcome_in_hospital_mortality"].sum() >= 10 + assert (~df["outcome_in_hospital_mortality"]).sum() >= 10 + assert df["outcome_discharge_disposition"].value_counts().min() >= 10 + assert df["outcome_severity_score"].nunique() > 10 + assert df[SURVIVAL_COLUMNS[-1]].sum() >= 10 # events within 5 years + + +def test_survival_horizons_are_cumulative(df): + flags = df[SURVIVAL_COLUMNS].to_numpy() + assert (np.diff(flags.astype(int), axis=1) >= 0).all() + + +def test_death_disposition_matches_mortality(df): + died = df["outcome_discharge_disposition"] == "death" + assert (died == df["outcome_in_hospital_mortality"]).all() + + +def test_deterministic_for_a_seed(): + pd.testing.assert_frame_equal(generate_dataset(50, 7), generate_dataset(50, 7)) + assert not generate_dataset(50, 7).equals(generate_dataset(50, 8)) + + +def test_written_dataset_roundtrip_and_metadata(tmp_path): + written = write_dataset(tmp_path, n_samples=N_SAMPLES, seed=0) + loaded = pd.read_parquet(tmp_path / "data.parquet") + pd.testing.assert_frame_equal(written, loaded) + + entry = json.loads((tmp_path / "metadata.json").read_text())["entries"][0] + stats = entry["datasetStats"] + assert [f["name"] for f in entry["features"]] == ALL_COLUMNS[:8] + assert [o["name"] for o in entry["outcomes"]] == ALL_COLUMNS[8:12] + assert set(stats["featureStats"]) == set(ALL_COLUMNS[:8]) + assert set(stats["outcomeStats"]) == set(ALL_COLUMNS[8:12]) + + age = stats["featureStats"]["patient_demographics_age"] + assert age["numOfNotNull"] == 100 + assert age["min"] == loaded["patient_demographics_age"].min() + assert age["q1"] <= age["q2"] <= age["q3"] + disp = stats["outcomeStats"]["outcome_discharge_disposition"] + assert sum(disp["cardinalityPerItem"].values()) == 100 + mortality = stats["outcomeStats"]["outcome_in_hospital_mortality"] + assert mortality["numOfTrue"] == loaded["outcome_in_hospital_mortality"].sum() + + +def test_flcore_dt4h_loader_reads_the_dataset(synthetic_data_dir): + """The generated files go through flcore's own loader without changes.""" + from flcore.datasets import load_dt4h + + config = { + "metadata_file": str(synthetic_data_dir / "metadata.json"), + "data_file": str(synthetic_data_dir / "data.parquet"), + "normalization_method": "IQR", + "train_size": 0.7, + "task": "classification", + "train_labels": [ + "patient_demographics_gender", + "patient_demographics_age", + "patient_flags_smoker", + ], + "target_labels": ["outcome_discharge_disposition"], + } + (X_train, y_train), (X_test, y_test) = load_dt4h(config) + assert len(X_train) == 70 and len(X_test) == 30 + assert not X_train.isna().any().any() + assert set(y_train) | set(y_test) == {0, 1, 2, 3} From c678ac43fbbbf37d71837775c93fd43c759c9a31 Mon Sep 17 00:00:00 2001 From: faildeny Date: Tue, 27 Oct 2026 16:45:02 +0100 Subject: [PATCH 2/5] Fix missing dependency for parquet files --- requirements.txt | 1 + 1 file changed, 1 insertion(+) diff --git a/requirements.txt b/requirements.txt index c34ac9b..dff296c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -16,3 +16,4 @@ pdfkit==1.0.0 scikit-survival==0.22.2 fpboost==0.1.0 setuptools==80.9.0 +pyarrow==25.0.1 From 057b89ae45a1a0f6114e33985d321b0d4e095938 Mon Sep 17 00:00:00 2001 From: faildeny Date: Tue, 27 Oct 2026 16:54:51 +0100 Subject: [PATCH 3/5] Add parallel test runs --- .github/workflows/tests.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 1e52602..8d31f9d 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -43,7 +43,7 @@ jobs: python client_cmd.py --help > /dev/null - name: Run tests - run: pytest tests --verbose --durations=10 + run: pytest tests -n 4 --verbose --durations=10 - name: Upload federation logs on failure if: failure() From 2c77e21d338e88fa816ad9197a8deef379bc8ab4 Mon Sep 17 00:00:00 2001 From: faildeny Date: Tue, 27 Oct 2026 16:59:38 +0100 Subject: [PATCH 4/5] Remove data tests --- tests/test_synthetic_data.py | 130 ----------------------------------- 1 file changed, 130 deletions(-) delete mode 100644 tests/test_synthetic_data.py diff --git a/tests/test_synthetic_data.py b/tests/test_synthetic_data.py deleted file mode 100644 index 4397a6b..0000000 --- a/tests/test_synthetic_data.py +++ /dev/null @@ -1,130 +0,0 @@ -"""The synthetic generator must reproduce the layout of the DT4H sample files.""" - -import json - -import numpy as np -import pandas as pd -import pytest -from synthetic_data import ( - ALL_COLUMNS, - DEFAULT_N_SAMPLES as N_SAMPLES, - DISPOSITIONS, - SURVIVAL_COLUMNS, - generate_dataset, - write_dataset, -) - -# Dtypes of synthetic_dt4h_dataset / synthetic_survival (strings are "object" -# or "str" depending on the pandas version, so they are compared by kind). -EXPECTED_KINDS = { - "patient_demographics_gender": "O", - "encounters_encounterClass": "O", - "encounters_admissionYear": "i", - "patient_demographics_age": "i", - "encounters_lengthOfStay": "i", - "vital_signs_weight_value_p6mo_min": "f", - "med_admins_history_diuretics_loop_any": "b", - "patient_flags_smoker": "b", - "outcome_in_hospital_mortality": "b", - "outcome_readmission_30d": "b", - "outcome_discharge_disposition": "O", - "outcome_severity_score": "f", - **{c: "b" for c in SURVIVAL_COLUMNS}, -} - - -@pytest.fixture(scope="module") -def df(): - return generate_dataset(N_SAMPLES, seed=0) - - -def test_has_100_rows_and_sample_file_columns(df): - assert len(df) == 100 - assert list(df.columns) == ALL_COLUMNS - assert len(ALL_COLUMNS) == 19 - - -def test_dtypes_match_sample_files(df): - for col, kind in EXPECTED_KINDS.items(): - assert df[col].dtype.kind == kind, f"{col}: {df[col].dtype}" - - -def test_no_missing_values(df): - assert not df.isna().any().any() - - -def test_value_ranges_match_metadata_sample(df): - assert df["patient_demographics_age"].between(18, 97).all() - assert df["encounters_lengthOfStay"].between(1, 45).all() - assert df["vital_signs_weight_value_p6mo_min"].between(42.5, 135.0).all() - assert df["outcome_severity_score"].between(0.0, 10.0).all() - assert set(df["outcome_discharge_disposition"]) == set(DISPOSITIONS) - - -def test_every_outcome_is_trainable_on_a_70_30_split(df): - # Each class needs enough members to land in both train and test splits. - assert df["outcome_in_hospital_mortality"].sum() >= 10 - assert (~df["outcome_in_hospital_mortality"]).sum() >= 10 - assert df["outcome_discharge_disposition"].value_counts().min() >= 10 - assert df["outcome_severity_score"].nunique() > 10 - assert df[SURVIVAL_COLUMNS[-1]].sum() >= 10 # events within 5 years - - -def test_survival_horizons_are_cumulative(df): - flags = df[SURVIVAL_COLUMNS].to_numpy() - assert (np.diff(flags.astype(int), axis=1) >= 0).all() - - -def test_death_disposition_matches_mortality(df): - died = df["outcome_discharge_disposition"] == "death" - assert (died == df["outcome_in_hospital_mortality"]).all() - - -def test_deterministic_for_a_seed(): - pd.testing.assert_frame_equal(generate_dataset(50, 7), generate_dataset(50, 7)) - assert not generate_dataset(50, 7).equals(generate_dataset(50, 8)) - - -def test_written_dataset_roundtrip_and_metadata(tmp_path): - written = write_dataset(tmp_path, n_samples=N_SAMPLES, seed=0) - loaded = pd.read_parquet(tmp_path / "data.parquet") - pd.testing.assert_frame_equal(written, loaded) - - entry = json.loads((tmp_path / "metadata.json").read_text())["entries"][0] - stats = entry["datasetStats"] - assert [f["name"] for f in entry["features"]] == ALL_COLUMNS[:8] - assert [o["name"] for o in entry["outcomes"]] == ALL_COLUMNS[8:12] - assert set(stats["featureStats"]) == set(ALL_COLUMNS[:8]) - assert set(stats["outcomeStats"]) == set(ALL_COLUMNS[8:12]) - - age = stats["featureStats"]["patient_demographics_age"] - assert age["numOfNotNull"] == 100 - assert age["min"] == loaded["patient_demographics_age"].min() - assert age["q1"] <= age["q2"] <= age["q3"] - disp = stats["outcomeStats"]["outcome_discharge_disposition"] - assert sum(disp["cardinalityPerItem"].values()) == 100 - mortality = stats["outcomeStats"]["outcome_in_hospital_mortality"] - assert mortality["numOfTrue"] == loaded["outcome_in_hospital_mortality"].sum() - - -def test_flcore_dt4h_loader_reads_the_dataset(synthetic_data_dir): - """The generated files go through flcore's own loader without changes.""" - from flcore.datasets import load_dt4h - - config = { - "metadata_file": str(synthetic_data_dir / "metadata.json"), - "data_file": str(synthetic_data_dir / "data.parquet"), - "normalization_method": "IQR", - "train_size": 0.7, - "task": "classification", - "train_labels": [ - "patient_demographics_gender", - "patient_demographics_age", - "patient_flags_smoker", - ], - "target_labels": ["outcome_discharge_disposition"], - } - (X_train, y_train), (X_test, y_test) = load_dt4h(config) - assert len(X_train) == 70 and len(X_test) == 30 - assert not X_train.isna().any().any() - assert set(y_train) | set(y_test) == {0, 1, 2, 3} From 0b12e0010c026b6c02c84d5517f61fec5c3892c7 Mon Sep 17 00:00:00 2001 From: Grzegorz Skorupko Date: Thu, 29 Oct 2026 15:49:28 +0100 Subject: [PATCH 5/5] Update Python version from 3.10 to 3.11 Signed-off-by: Grzegorz Skorupko --- .github/workflows/tests.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 8d31f9d..e64a3fa 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -20,10 +20,10 @@ jobs: - name: Checkout uses: actions/checkout@v4 - - name: Set up Python 3.10 + - name: Set up Python 3.11 uses: actions/setup-python@v5 with: - python-version: "3.10" + python-version: "3.11" cache: pip cache-dependency-path: | requirements.txt