Spaces:
Paused
Paused
File size: 2,319 Bytes
e40dce0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | 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'}
|