import unittest from datetime import date, timedelta import numpy as np from forecasting import (TransitionModel, build_forecast, e1rm, profile, sessions, training_blocks) from utils import get_exercise_graph_model, _format_kg def rows(pairs, start=date(2026, 1, 1), interval=7): return tuple((start + timedelta(days=i * interval), w, r) for i, (w,r) in enumerate(pairs)) def forecast(data, weeks=8): return build_forecast([], [r[0] for r in data], weeks, weights=[r[1] for r in data], repetitions=[r[2] for r in data]) class ForecastTests(unittest.TestCase): def setUp(self): self.history = rows([(70,7),(70,8),(70,9),(70.5,7),(70.5,8),(70.5,9), (71,5),(71,6),(71,6),(71,6),(71,6),(71,7)]) def test_fractional_loads_round_trip(self): self.assertAlmostEqual(float(e1rm(93.5,7)),113.1942,places=3) f=forecast(self.history) for point in f['points']: self.assertEqual(point['value'],round(float(e1rm(point['projected_weight'], point['projected_repetitions'])),1)) self.assertEqual(_format_kg(96.75),'96.75') def test_stalls_do_not_force_weekly_load_increases(self): m=TransitionModel(self.history) self.assertEqual(m.profile['block_stalls'],3) self.assertEqual(m.profile['ceiling'],9) self.assertEqual(m.next_set()[0][0],(71,7)) w,r,p=m.outcomes([71],[7],[0]) self.assertTrue(np.all(p[w>71]==0)) def test_loading_at_ceiling_uses_recorded_step(self): m=TransitionModel(self.history) w,r,p=m.outcomes([71],[9],[0]) self.assertTrue(np.any((w>71)&(p>0))) self.assertLessEqual(float(w.max()),71.5) def test_plateau_does_not_invent_gains(self): f=forecast(rows([(50,8)]*16),52) self.assertTrue(all(p['projected_weight']==50 and p['projected_repetitions']==8 for p in f['points'])) def test_mixed_observed_load_steps_are_preserved(self): m = TransitionModel(rows([(100,7),(100.5,7),(101.5,7),(102,7),(104,7)])) w, r, p = m.outcomes([104], [7], [0]) increases = set(np.round(w[(w > 104) & (p > 0)] - 104, 3)) self.assertEqual(increases, {0.5, 1.0, 2.0}) def test_ceiling_does_not_turn_rep_gains_into_stall_evidence(self): m = TransitionModel(rows([(100,6),(100,7),(101,6),(101,7),(102,6),(102,7)])) w, r, p = m.outcomes([102], [7], [0]) # At the ceiling only the explicit prior can repeat this set; # the observed rep-building transitions contribute no probability. self.assertAlmostEqual(float(p[(w == 102) & (r == 7)].sum()), float(p[0,0])) self.assertTrue(np.any((w > 102) & (p > 0))) w, r, p = m.outcomes([102], [6], [0]) self.assertTrue(np.any((w == 102) & (r == 7) & (p > 0))) def test_large_load_jump_is_not_reused_as_routine_progression(self): m = TransitionModel(rows([(50,7),(50.5,7),(51,7),(100,7)])) w, r, p = m.outcomes([100], [7], [0]) self.assertEqual(set(w[(w > 100) & (p > 0)]), {100.5}) def test_months_off_excludes_old_performance(self): older=rows([(150,10)]*12,start=date(2024,1,1)) fresh=rows([(50,6),(50,7),(50,7),(50,8)], start=date(2026,1,1)) f=forecast(older+fresh) expected=forecast(fresh) self.assertEqual(f['points'],expected['points']) self.assertEqual(f['training_blocks'],2) self.assertEqual(f['observed_points'],4) self.assertFalse(forecast(older+fresh[:3])['available']) def test_block_detection_is_prefix_invariant(self): data=self.history+rows([(60,8)]*10,start=date(2027,1,1)) full=training_blocks(data) for cutoff in range(1,len(data)+1): trimmed=tuple(tuple(r for r in b if r[0]<=data[cutoff-1][0]) for b in full if b[0][0]<=data[cutoff-1][0]) self.assertEqual(training_blocks(data[:cutoff]),trimmed) def test_duplicate_date_uses_one_coherent_best_set(self): extra=(self.history[-1][0],65,5) data=self.history+(self.history[-1],extra) self.assertEqual(forecast(data)['points'],forecast(self.history)['points']) def test_nonfinite_and_unsupported_data(self): invalid=rows([(float('nan'),8),(30,40),(0,6),(float('inf'),6)]) self.assertFalse(forecast(invalid)['available']) def test_observed_anchor_and_consistent_horizons(self): small,large=forecast(self.history,4),forecast(self.history,52) self.assertEqual(small['points'][0]['projected_weight'],71) self.assertEqual(small['points'][0]['projected_repetitions'],7) self.assertEqual(small['next_session'],large['next_session']) for a,b in zip(small['session_points'],large['session_points']): self.assertEqual(a,b) def test_interval_order_and_cadence(self): data=rows([(50,6),(50,7),(50,7),(50,8),(50,8),(50,9)],interval=14) f=forecast(data,8) self.assertEqual(f['profile']['cadence_days'],14) self.assertEqual(len(f['session_points']),4) self.assertEqual(f['next_session']['date'],data[-1][0]+timedelta(days=14)) for point in f['points']: self.assertLessEqual(point['lower'],point['value']) self.assertLessEqual(point['value'],point['upper']) def test_cached_forecast_is_not_mutated_by_renderer(self): f=forecast(self.history) f['points'][0]['value']=-1 self.assertGreater(forecast(self.history)['points'][0]['value'],0) def test_display_window_does_not_change_training_or_seed_reps(self): def graph(visible, enabled=True): return get_exercise_graph_model('Test', [round(float(e1rm(w,r)),1) for _,w,r in visible], [r for _,_,r in visible], [w for _,w,_ in visible], [d for d,_,_ in visible], ['']*len(visible), 'All',1,1,forecast_enabled=enabled,forecast_training_rows=[dict(start_date=d,weight=w,repetitions=r) for d,w,r in self.history]) full,short=graph(self.history),graph(self.history[-3:]) # Equivalent rep tables intentionally reflect the visible rep range. for a,b in zip(full['forecast']['points'],short['forecast']['points']): self.assertEqual({k:v for k,v in a.items() if k!='top_sets'}, {k:v for k,v in b.items() if k!='top_sets'}) self.assertEqual(full['forecast']['target_weight_increment'],short['forecast']['target_weight_increment']) self.assertIsNone(graph(self.history,False)['forecast_target_reps']) hits=full['hover_details'] for a,b in zip(hits,hits[1:]): self.assertEqual(a['hit_end'],b['hit_start']) self.assertLess(a['hit_start'],a['hit_end']) if __name__=='__main__': unittest.main()