35 lines
1.7 KiB
Python
35 lines
1.7 KiB
Python
import unittest
|
|
|
|
from evaluate_streaming import classification, last_available
|
|
|
|
|
|
class CausalStreamingScores(unittest.TestCase):
|
|
def test_startup_future_and_stale_outputs_are_unavailable(self):
|
|
points = [[100, .2], [200, .9], [900, .3]]
|
|
self.assertIsNone(last_available(points, 99, 200))
|
|
self.assertEqual(last_available(points, 150, 200), [100, .2])
|
|
self.assertEqual(last_available(points, 200, 200), [200, .9])
|
|
self.assertEqual(last_available(points, 400, 200), [200, .9])
|
|
self.assertIsNone(last_available(points, 401, 200))
|
|
self.assertEqual(last_available(points, 1000, 200), [900, .3])
|
|
|
|
def test_missing_outputs_do_not_count_as_correct_negative_predictions(self):
|
|
rows = [dict(can_reply_now='0', status='unavailable', score=''),
|
|
dict(can_reply_now='0', status='available', score=.7),
|
|
dict(can_reply_now='1', status='available', score=.8),
|
|
dict(can_reply_now='', status='available', score=.1)]
|
|
result = classification(rows)
|
|
self.assertEqual((result['n'], result['requested_labeled'], result['unavailable_labeled']), (2, 3, 1))
|
|
self.assertEqual((result['tn'], result['fp'], result['fn'], result['tp']), (0, 1, 0, 1))
|
|
self.assertEqual(result['accuracy'], .5)
|
|
|
|
def test_threshold_is_strict_and_empty_scores_have_no_accuracy(self):
|
|
self.assertEqual(classification([dict(can_reply_now='1',status='available',score=.5)])['fn'], 1)
|
|
result = classification([dict(can_reply_now='0',status='unavailable',score='')])
|
|
self.assertEqual(result['n'], 0)
|
|
self.assertIsNone(result['accuracy'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|