tests/test_randomization_db.py

import os
import sqlite3
import subprocess
import time
from pathlib import Path

import pytest


PROJECT_DIR = Path(__file__).resolve().parents[1]
OTREE_BIN = PROJECT_DIR.parents[1] / 'venv' / 'bin' / 'otree'


def _run_otree_and_return_db(session: str, n: int) -> Path:
    env = os.environ.copy()
    db_path = PROJECT_DIR / f"tmp_rand_{session}_{n}_{int(time.time()*1000)}.sqlite3"
    env['OTREE_DATABASE_URL'] = f"sqlite:///{db_path}"
    # Remove default project DB to avoid accidental reuse
    default_db = PROJECT_DIR / 'db.sqlite3'
    if default_db.exists():
        default_db.unlink()
    proc = subprocess.run(
        [str(OTREE_BIN), 'test', session, str(n)],
        cwd=str(PROJECT_DIR),
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        env=env,
    )
    if proc.returncode != 0:
        print(proc.stdout)
    assert proc.returncode == 0, f"oTree test failed for {session} N={n}"
    return db_path


def _fetch_orders(db_path: Path):
    con = sqlite3.connect(str(db_path))
    try:
        cur = con.cursor()
        cur.execute(
            "SELECT order_a, order_b, order_c FROM policy_survey_biases_player WHERE round_number=1"
        )
        rows = cur.fetchall()
        a = [r[0] for r in rows]
        b = [r[1] for r in rows]
        c = [r[2] for r in rows]
        return a, b, c
    finally:
        con.close()


@pytest.mark.parametrize('n', [2, 3, 6])
def test_persisted_orders_balanced_per_session(n):
    db = _run_otree_and_return_db('survey_biases_full', n)
    a, b, c = _fetch_orders(db)
    for arr in (a, b, c):
        counts = {v: arr.count(v) for v in set(arr)}
        assert len(counts) == 2, f"Expected 2 conditions, got {counts}"
        diff = abs(list(counts.values())[0] - list(counts.values())[1])
        assert diff <= 1, (n, arr, counts)


def test_persisted_orders_vary_across_runs():
    seen = dict(a=set(), b=set(), c=set())
    n = 6
    for _ in range(3):
        db = _run_otree_and_return_db('survey_biases_full', n)
        a, b, c = _fetch_orders(db)
        seen['a'].update(a)
        seen['b'].update(b)
        seen['c'].update(c)
    assert seen['a'] == {'info_before', 'info_after'}
    assert seen['b'] == {'support_first', 'eval_first'}
    assert seen['c'] == {'recall_first', 'opinion_first'}