1  from pprint import pprint  # noqa
 2  
 3  from b4mad_racing_website.models import Copilot
 4  from django.test import TransactionTestCase
 5  
 6  from telemetry.models import Coach, Driver
 7  from telemetry.pitcrew.coach_copilots import CoachCopilots
 8  from telemetry.pitcrew.history import History
 9  
10  from .utils import get_session_df, read_responses, save_responses  # noqa
11  
12  
13  class TestDebugApp(TransactionTestCase):
14      fixtures = [
15          "game.json",
16          "track.json",
17          "car.json",
18          "session.json",
19          "sessiontype.json",
20          "lap.json",
21          "fastlap.json",
22          "fastlapsegment.json",
23          "driver.json",
24          "coach.json",
25          "trackguide.json",
26          "trackguidenote.json",
27          "landmark.json",
28          "copilot.json",
29          "copilotinstance.json",
30          "profile.json",
31          "user.json",
32      ]
33      maxDiff = None
34  
35      do_save_responses = True
36  
37      def test_brake(self):
38          # iRacing / Mazda MX-5 Cup / okayama short
39  
40          session_id = "1694266648"
41          driver = Driver.objects.get(name="durandom")
42          coach = driver.coach
43          coach.mode = Coach.MODE_COPILOTS
44          coach.save()
45  
46          # delete all Copilots where the slug is not "brake"
47          Copilot.objects.exclude(slug="braker").delete()
48  
49          history = History()
50  
51          coach = CoachCopilots(history, coach)
52  
53          session_df = get_session_df(session_id, measurement="laps_cc", bucket="racing")
54  
55          row = session_df.iloc[0].to_dict()
56          topic = row["topic"].replace("Jim", "durandom")
57          coach.notify(topic, row)
58          history.init()
59          history._do_init = False
60  
61          captured_responses = []
62          try:
63              for index, row in session_df.iterrows():
64                  row = row.to_dict()
65                  response = coach.notify(topic, row, row["_time"])
66                  if response:
67                      captured_responses.append((row["DistanceRoundTrack"], response))
68          except Exception as e:
69              raise e
70          finally:
71              print("stopping history thread")
72              history.disconnect()
73  
74          responses_file = "test_copilot_brake"
75          expected_responses = read_responses(responses_file)
76          if self.do_save_responses:
77              save_responses(captured_responses, responses_file)
78  
79          # pprint(captured_responses, width=200)
80          self.assertEqual(captured_responses, expected_responses)