|
- """test_adjusters.py -- test pipeline Adjusters (eg TargetTimeAdjuster)"""
- import unittest
- import unittest.mock as mock
- import pipeline.adjusters as adjusters
-
- from test.mocks import MockFeature, MockInterval
-
- class TestAdjuster(unittest.TestCase):
- """Test the generic Adjuster class"""
-
- def test_init(self):
- """Test the Adjuster can be initialised"""
- adjuster = adjusters.Adjuster()
- self.assertEqual(adjuster.features, [])
-
- def test_adjust(self):
- """Test the generic adjust"""
- adjuster = adjusters.Adjuster()
- self.assertEqual(adjuster.adjust(), [])
- self.assertEqual(adjuster.features, [])
-
- class TestTargetTimeAdjuster(unittest.TestCase):
- """Test the TargetTimeAdjuster
-
- TTA drops Features until the target time is reached (or within a margin)"""
-
- def test_init(self):
- """Test the TTA can be initialised"""
- tta = adjusters.TargetTimeAdjuster()
- self.assertEqual(tta.features, [])
-
- def test_features_total_time(self):
- """Test the TTA can calculate the total time of Features
-
- Test:
- - input duration floats: 1.0, 2.0, 3.0, 4.0 == 10.0
- """
- tta = adjusters.TargetTimeAdjuster()
- features = []
- for i in range(1, 5):
- features.append(make_feature(duration=i*1.0))
-
- self.assertEqual(tta._features_total_time(features), 10.0)
- self.assertEqual(tta._features_total_time([]), 0.0)
- self.assertIs(type(tta._features_total_time([])), float)
-
- def test_determine_margin(self):
- """Test the TTA can determine the target time margins
-
- Args: time, margin, strategy (strategy in: ABSOLUTE, PERCENT)
-
- Test:
- - margin of zero
- - margin of 5.0
- - margin of 10.0
- - margin of 100.0
- - both ABSOLUTE and PERCENT strategies
-
- TODO: figure out what should be done with negative margins & margins > 100.0
- """
-
- tta = adjusters.TargetTimeAdjuster()
- with self.subTest("ABSOLUTE"):
- strategy = adjusters.TargetTimeAdjuster._STRATEGY.ABSOLUTE
- test_cases = []
- # populate test cases with tuples of (time, margin, expected)
- # zero margin
- test_cases.append((60.0, 0.0, (60.0, 60.0)))
- # margin of 5.0
- test_cases.append((60.0, 5.0, (55.0, 65.0)))
- # margin of 10.0
- test_cases.append((60.0, 10.0, (50.0, 70.0)))
- # margin of 100.0
- test_cases.append((60.0, 100.0, (0.0, 160.0)))
-
- # test
- for time, margin, expected in test_cases:
- self.assertEqual(tta._determine_margin(time, margin, strategy), expected)
-
- with self.subTest("PERCENT"):
- strategy = adjusters.TargetTimeAdjuster._STRATEGY.PERCENT
- test_cases = []
- # populate test cases with tuples of (time, margin, expected) as above
- # zero margin
- test_cases.append((60.0, 0.0, (60.0, 60.0)))
- # margin of 5.0
- test_cases.append((60.0, 5.0, (57.0, 63.0)))
- # margin of 10.0
- test_cases.append((60.0, 10.0, (54.0, 66.0)))
- # margin of 100.0
- test_cases.append((60.0, 100.0, (0.0, 120.0)))
-
- # test
- for time, margin, expected in test_cases:
- self.assertEqual(tta._determine_margin(time, margin, strategy), expected)
-
- def test_adjust_no_change(self):
- """Test adjusting of list of Features using TTA -- no change to list of Features
-
- Cases:
- - no Features --> []
- - [Features] with total time < target time --> unchanged list
- - [Features] with total time = target time --> unchanged list
-
- TODO: test with Features > target
- """
- with self.subTest("no Features"):
- tta = adjusters.TargetTimeAdjuster()
- self.assertEqual(tta.adjust(), [])
-
- with self.subTest("Features < target time"):
- features = []
- for i in range(1, 5):
- features.append(make_feature(duration=i*1.0))
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=20.0)
- self.assertEqual(tta.adjust(), features)
-
- with self.subTest("Features = target time"):
- features = []
- for i in range(1, 5):
- features.append(make_feature(duration=i*1.0))
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=10.0)
- self.assertEqual(tta.adjust(), features)
-
-
- def test_sort_by_score_time(self):
- """Test sorting of list of Features by score (primary) and time (secondary)
-
- Cases:
- - [(15.0, 1.0), (10.0, 1.0), (12.0, 1.0)] --> [(10.0, 1.0), (12.0, 1.0), (15.0, 1.0)] # score equal, sort by time
- - [(15.0, 1.0), (10.0, 4.0), (12.0, 3.0)] --> [(15.0, 1.0), (12.0, 3.0), (10.0, 4.0)] # sort by score
- - [(15.0, 1.0), (10.0, 1.0), (12.0, 2.0)] --> [(10.0, 1.0), (15.0, 1.0), (12.0, 2.0)] # mixed: scores below duration
- - [] --> []
- - [(15.0, 1.0)] --> [(15.0, 1.0)]
-
- Cases giving RDH trouble:
- - [(16.0, 1.0), (16.0, 1.0), (1.0, 1.0), (1.0, 1.0)] --> [(1.0, 1.0), (1.0, 1.0), (16.0, 1.0), (16.0, 1.0)] # multiple lowest scoring, multiple shortest duration
- """
-
- tta = adjusters.TargetTimeAdjuster()
- with self.subTest("score equal, sort by duration"):
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=1.0),
- make_feature(duration=12.0, score=1.0)
- ]
- self.assertEqual(tta._sort_by_score_time(features), [features[1], features[2], features[0]])
-
- with self.subTest("sort by score, duration irrelevant"):
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=4.0),
- make_feature(duration=12.0, score=3.0)
- ]
- self.assertEqual(tta._sort_by_score_time(features), [features[0], features[2], features[1]])
-
- with self.subTest("mixed: scores below duration"):
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=1.0),
- make_feature(duration=12.0, score=2.0)
- ]
- self.assertEqual(tta._sort_by_score_time(features), [features[1], features[0], features[2]])
-
- with self.subTest("empty"):
- self.assertEqual(tta._sort_by_score_time([]), [])
-
- with self.subTest("single"):
- features = [mock.Mock(duration=15.0, score=1.0)]
- self.assertEqual(tta._sort_by_score_time(features), features)
-
- with self.subTest("multiple lowest scoring, multiple shortest duration"):
- features = [
- make_feature(duration=16.0, score=1.0),
- make_feature(duration=16.0, score=1.0),
- make_feature(duration=1.0, score=1.0),
- make_feature(duration=1.0, score=1.0)
- ]
- self.assertEqual(tta._sort_by_score_time(features), [features[2], features[3], features[0], features[1]])
-
-
- def test_adjust_changes(self):
- """Test adjusting of list of Features using TTA -- changes to list of Features
-
- All cases have total time > target time.
-
- In the cases, specification is Feature(duration, score)
- Cases:
- - target = 30.0, margin = 0.0
- + [(15.0, 1.0), (10.0, 1.0), (12.0, 1.0)] --> [(15.0, 1.0), (12.0, 1.0)] # scores equal, drop smallest
- + [(15.0, 2.0), (10.0, 2.0), (12.0, 1.0)] --> [(15.0, 1.0), (10.0, 1.0)] # drop lowest scoring (1)
- + [(15.0, 1.0), (10.0, 1.0), (12.0, 2.0)] --> [(15.0, 1.0), (12.0, 2.0)] # drop lowest scoring (2)
-
- - target = 30.0, margin = 4.0
- + [(15.0, 1.0), (10.0, 2.0), (12.0, 1.0)] --> [(15.0, 1.0), (12.0, 1.0)] # not lowest scoring, but within margin
- + [(16.0, 1.0), (16.0, 1.0), (1.0, 1.0), (1.0, 1.0)] --> [(16.0, 1.0), (16.0, 1.0)] # drop multiple lowest scoring, shortest duration
-
- """
- # target 30.0, margin 0.0 cases
- target, margin = 30.0, 0.0
-
- with self.subTest(f"target {target} margin {margin}"):
- with self.subTest("scores equal"):
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=1.0),
- make_feature(duration=12.0, score=1.0)
- ]
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=target, margin=margin)
- expected = [features[0], features[2]]
- output = tta.adjust()
- self.assertEqual(len(output), 2)
- self.assertEqual(output, expected)
- self.assertEqual(tta.features, expected)
-
- with self.subTest("drop lowest scoring (1)"):
- features = [
- make_feature(duration=15.0, score=2.0),
- make_feature(duration=10.0, score=2.0),
- make_feature(duration=12.0, score=1.0)
- ]
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=target, margin=margin)
- expected = [features[0], features[1]]
- output = tta.adjust()
- self.assertEqual(len(output), 2)
- self.assertEqual(output, expected)
- self.assertEqual(tta.features, expected)
-
- with self.subTest("drop lowest scoring (2)"):
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=1.0),
- make_feature(duration=12.0, score=2.0)
- ]
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=target, margin=margin)
- expected = [features[0], features[2]]
- output = tta.adjust()
- self.assertEqual(len(output), 2)
- self.assertEqual(output, expected)
- self.assertEqual(tta.features, expected)
-
- # target 30.0, margin 4.0 cases
- target, margin, strategy = 30.0, 4.0, adjusters.TargetTimeAdjuster._STRATEGY.ABSOLUTE
-
- with self.subTest(f"target {target} margin {margin}"):
- with self.subTest("not lowest scoring, but within margin"):
- # explanation: dropping the 10.0 feature would put us at 27.0, which is within the margin (26.0, 34.0)
- features = [
- make_feature(duration=15.0, score=1.0),
- make_feature(duration=10.0, score=2.0),
- make_feature(duration=12.0, score=1.0)
- ]
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=target,
- margin=margin, strategy=strategy)
- expected = [features[0], features[2]]
- output = tta.adjust()
- self.assertEqual(len(output), 2)
- self.assertEqual(output, expected)
- self.assertEqual(tta.features, expected)
-
- with self.subTest("drop multiple lowest scoring, shortest duration"):
- # explanation: dropping the 1.0 features would put us at 32.0, which is within the margin (26.0, 34.0)
- features = [
- make_feature(duration=16.0, score=1.0),
- make_feature(duration=16.0, score=1.0),
- make_feature(duration=1.0, score=1.0),
- make_feature(duration=1.0, score=1.0),
- make_feature(duration=1.0, score=1.0),
- make_feature(duration=1.0, score=1.0)
- ]
- tta = adjusters.TargetTimeAdjuster(features=features, target_time=target,
- margin=margin, strategy=strategy)
- expected = [features[0], features[1], features[2], features[3]]
- output = tta.adjust()
- self.assertEqual(len(output), 4)
- self.assertEqual(output, expected)
- self.assertEqual(tta.features, expected)
-
-
- def make_feature(duration, score=1.0):
- """Helper function to create a MockFeature from duration and score"""
- return MockFeature(interval=MockInterval.from_duration(duration), score=score)
|