blob: 3000a8c90933e42d7ca532535ae6cbe6c5471d79 [file] [log] [blame] [edit]
#!/usr/bin/env python3
# Copyright 2022 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Test and benchmark the experiment module."""
import pathlib
import tempfile
import textwrap
import unittest
from experiment import *
import fpsutils
import pandas as pd
import simulate_fpstudy
class Test_has_user_groups(unittest.TestCase):
"""Test the `has_user_groups` method."""
BARE_DECISIONS_TABLE = pd.DataFrame(
columns=Experiment.DECISION_TABLE_COLS,
)
BARE_DECISIONS_TABLE_WITH_GROUPS = pd.DataFrame(
columns=Experiment.DECISION_TABLE_COLS
+ Experiment.DECISION_TABLE_GROUP_COLS,
)
def setUp(self) -> None:
"""Run before each test."""
self.exp = Experiment(
num_users=72,
num_fingers=3,
num_verification=40,
)
def test_without_any_tables(self):
self.assertFalse(self.exp.has_user_groups())
def test_with_far_no_user_groups(self):
self.exp.add_far_decisions(self.BARE_DECISIONS_TABLE.copy())
self.assertFalse(self.exp.has_user_groups())
def test_with_far_and_user_groups(self):
self.exp.add_far_decisions(self.BARE_DECISIONS_TABLE_WITH_GROUPS.copy())
self.assertTrue(self.exp.has_user_groups())
def test_with_frr_no_user_groups(self):
self.exp.add_frr_decisions(self.BARE_DECISIONS_TABLE.copy())
self.assertFalse(self.exp.has_user_groups())
def test_with_frr_and_user_groups(self):
self.exp.add_frr_decisions(self.BARE_DECISIONS_TABLE_WITH_GROUPS.copy())
self.assertTrue(self.exp.has_user_groups())
class Test_fa_query(unittest.TestCase):
def setUp(self):
# VerifyUser VerifyFinger VerifySample EnrollUser EnrollFinger Decision
# This can take about 2 seconds.
self.far_decisions = simulate_fpstudy.GenerateFARResults(
num_users=72,
num_fingers=3,
num_verify_samples=40,
user_groups=None,
verbose=True,
)
self.exp = Experiment(
num_users=72,
num_fingers=3,
num_verification=40,
far_decisions=self.far_decisions,
)
def test_benchmark_fa_query1(self):
print("Query 1 Timing")
fpsutils.benchmark(
"self.exp.fa_query(verify_user_id=10071)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query(verify_user_id=10071, verify_finger_id=0)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60, enroll_user_id=10057)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60, enroll_user_id=10057, enroll_finger_id=4)",
globals={**globals(), **locals()},
)
def test_benchmark_fa_query2(self):
print("Query 2 Timing")
fpsutils.benchmark(
"self.exp.fa_query2(verify_user_id=10071)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query2(verify_user_id=10071, verify_finger_id=0)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query2(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query2(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60, enroll_user_id=10057)",
globals={**globals(), **locals()},
)
fpsutils.benchmark(
"self.exp.fa_query2(verify_user_id=10071, verify_finger_id=0, verify_sample_index=60, enroll_user_id=10057, enroll_finger_id=4)",
globals={**globals(), **locals()},
)
class Test_Experiment_CSV(unittest.TestCase):
"""Test the CSV reading and writing capabilities of `Experiment`."""
CSV_FAR = textwrap.dedent(
"""\
EnrollUser,EnrollFinger,VerifyUser,VerifyFinger,VerifySample,Decision
10001,1,10002,3,12,REJECT
10001,1,10002,3,13,REJECT
"""
)
CSV_FAR_DATAFRAME = pd.DataFrame(
{
Experiment.TableCol.Enroll_User.value: [10001, 10001],
Experiment.TableCol.Enroll_Finger.value: [1, 1],
Experiment.TableCol.Verify_User.value: [10002, 10002],
Experiment.TableCol.Verify_Finger.value: [3, 3],
Experiment.TableCol.Verify_Sample.value: [12, 13],
Experiment.TableCol.Decision.value: [
Experiment.Decision.Reject.value,
Experiment.Decision.Reject.value,
],
}
)
CSV_FRR = textwrap.dedent(
"""\
EnrollUser,EnrollFinger,VerifyUser,VerifyFinger,VerifySample,Decision
10001,1,10001,1,25,ACCEPT
10001,2,10001,2,25,ACCEPT
"""
)
CSV_FRR_DATAFRAME = pd.DataFrame(
{
Experiment.TableCol.Enroll_User.value: [10001, 10001],
Experiment.TableCol.Enroll_Finger.value: [1, 2],
Experiment.TableCol.Verify_User.value: [10001, 10001],
Experiment.TableCol.Verify_Finger.value: [1, 2],
Experiment.TableCol.Verify_Sample.value: [25, 25],
Experiment.TableCol.Decision.value: [
Experiment.Decision.Accept.value,
Experiment.Decision.Accept.value,
],
}
)
CSV_USER_GROUP = textwrap.dedent(
"""\
User,Group
10001,A
10002,B
"""
)
CSV_USER_GROUP_DATAFRAME = pd.DataFrame(
{
Experiment.TableCol.User.value: [10001, 10002],
Experiment.TableCol.Group.value: ["A", "B"],
}
)
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.temp_csv = pathlib.Path(self.temp_dir.name) / "test.csv"
def tearDown(self) -> None:
self.temp_dir.cleanup()
return super().tearDown()
def test_read_far(self):
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
self.temp_csv.write_text(self.CSV_FAR)
exp.add_far_decisions_from_csv(self.temp_csv)
self.assertTrue(exp.has_far_decisions())
self.assertFalse(exp.has_frr_decisions())
df = exp.far_decisions()
self.assertTrue(df.equals(self.CSV_FAR_DATAFRAME))
def test_read_frr(self):
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
self.temp_csv.write_text(self.CSV_FRR)
exp.add_frr_decisions_from_csv(self.temp_csv)
self.assertFalse(exp.has_far_decisions())
self.assertTrue(exp.has_frr_decisions())
df = exp.frr_decisions()
self.assertTrue(df.equals(self.CSV_FRR_DATAFRAME))
def test_write_far(self):
exp = Experiment(
num_verification=0,
num_fingers=0,
num_users=0,
far_decisions=self.CSV_FAR_DATAFRAME,
)
self.temp_csv.unlink(missing_ok=True)
exp.far_decisions_to_csv(self.temp_csv)
self.assertEqual(self.temp_csv.read_text(), self.CSV_FAR)
def test_write_frr(self):
exp = Experiment(
num_verification=0,
num_fingers=0,
num_users=0,
frr_decisions=self.CSV_FRR_DATAFRAME,
)
self.temp_csv.unlink(missing_ok=True)
exp.frr_decisions_to_csv(self.temp_csv)
self.assertEqual(self.temp_csv.read_text(), self.CSV_FRR)
def test_read_user_group(self):
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
self.temp_csv.write_text(self.CSV_USER_GROUP)
exp.add_groups_from_csv(self.temp_csv)
self.assertTrue(exp.has_user_groups())
df = exp.user_groups_table()
self.assertTrue(df.equals(self.CSV_USER_GROUP_DATAFRAME))
def test_write_user_group(self):
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
exp.add_groups(self.CSV_USER_GROUP_DATAFRAME)
self.assertTrue(exp.has_user_groups())
self.temp_csv.unlink(missing_ok=True)
exp.user_groups_table_to_csv(self.temp_csv)
self.assertEqual(self.temp_csv.read_text(), self.CSV_USER_GROUP)
class Test_Experiment_User_Groups(unittest.TestCase):
"""Test the User Group capabilities of `Experiment`."""
def setUp(self) -> None:
self.temp_dir = tempfile.TemporaryDirectory()
self.temp_csv = pathlib.Path(self.temp_dir.name) / "test.csv"
def tearDown(self) -> None:
self.temp_dir.cleanup()
return super().tearDown()
FRR_DATAFRAME_GROUPS = pd.DataFrame(
{
Experiment.TableCol.Enroll_User.value: [10001, 10002],
Experiment.TableCol.Enroll_Finger.value: [1, 2],
Experiment.TableCol.Verify_User.value: [10001, 10002],
Experiment.TableCol.Verify_Finger.value: [1, 2],
Experiment.TableCol.Verify_Sample.value: [25, 25],
Experiment.TableCol.Decision.value: [
Experiment.Decision.Accept.value,
Experiment.Decision.Accept.value,
],
# With Enroll and Verify groups.
Experiment.TableCol.Enroll_Group.value: ["A", "B"],
Experiment.TableCol.Verify_Group.value: ["A", "B"],
}
)
FAR_DATAFRAME_GROUPS = pd.DataFrame(
{
Experiment.TableCol.Enroll_User.value: [10001, 10001],
Experiment.TableCol.Enroll_Finger.value: [1, 1],
Experiment.TableCol.Verify_User.value: [10002, 10002],
Experiment.TableCol.Verify_Finger.value: [3, 3],
Experiment.TableCol.Verify_Sample.value: [12, 13],
Experiment.TableCol.Decision.value: [
Experiment.Decision.Reject.value,
Experiment.Decision.Reject.value,
],
# With Enroll and Verify groups.
Experiment.TableCol.Enroll_Group.value: ["A", "A"],
Experiment.TableCol.Verify_Group.value: ["B", "B"],
}
)
"""This example requires the groups to be scanned from both enroll and verify."""
USER_GROUPS_TABLE = pd.DataFrame(
{
Experiment.TableCol.User.value: [10001, 10002],
Experiment.TableCol.Group.value: ["A", "B"],
}
)
def test_infer_groups_from_frr(self):
"""Test whether the user_groups can be scanned from the FRR table."""
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
exp.add_frr_decisions(self.FRR_DATAFRAME_GROUPS)
self.assertTrue(exp.has_user_groups())
user_groups = exp.user_groups_table()
self.assertIsNotNone(user_groups)
self.assertTrue(user_groups.equals(self.USER_GROUPS_TABLE))
def test_infer_groups_from_far(self):
"""Test whether the user_groups can be scanned from the FAR table.
Additionally, this checks whether groups are detected from both enroll
and verify columns.
"""
exp = Experiment(num_verification=0, num_fingers=0, num_users=0)
exp.add_far_decisions(self.FAR_DATAFRAME_GROUPS)
self.assertTrue(exp.has_user_groups())
user_groups = exp.user_groups_table()
self.assertIsNotNone(user_groups)
self.assertTrue(user_groups.equals(self.USER_GROUPS_TABLE))
if __name__ == "__main__":
unittest.main()