ya2/utils/asserts.py

   1 import threading
   2 from re import search
   3
   4
   5 class Assert:
   6
   7     @staticmethod
   8     def assert_render3d():
   9         preserve = ['camera', 'DIRECT']
  10         unexpected_nodes = [c for c in render.children if c.name not in preserve]
  11         for node in unexpected_nodes: Assert.__process_render3d_exception(node)
  12
  13     @staticmethod
  14     def __process_render3d_exception(node):
  15         render.ls()
  16         message = f'unexpected render3d node: {node.name}'
  17         raise Exception(message)
  18
  19     @staticmethod
  20     def assert_aspect2d():
  21         preserve = [
  22             'a2dBackground', 'a2dTopCenter', 'a2dTopCenterNS',
  23             'a2dBottomCenter', 'a2dBottomCenterNS', 'a2dLeftCenter',
  24             'a2dLeftCenterNS', 'a2dRightCenter', 'a2dRightCenterNS',
  25             'a2dTopLeft', 'a2dTopLeftNS', 'a2dTopRight',
  26             'a2dTopRightNS', 'a2dBottomLeft', 'a2dBottomLeftNS',
  27             'a2dBottomRight', 'a2dBottomRightNS', 'test_txt']
  28         unexpected_nodes = [c for c in aspect2d.children
  29                             if c.name not in preserve and not c.has_python_tag('preserve')]
  30         for node in unexpected_nodes: Assert.__process_aspect2d_exception(node)
  31
  32     @staticmethod
  33     def __process_aspect2d_exception(node):
  34         aspect2d.ls()
  35         message = f'unexpected aspect2d node: {node.name}'
  36         raise Exception(message)
  37
  38     @staticmethod
  39     def assert_render2d():
  40         preserve = ['aspect2d', 'pixel2d', 'camera2d']
  41         unexpected_nodes = [c for c in render2d.children if c.name not in preserve and not c.has_python_tag('preserve')]
  42         for node in unexpected_nodes: Assert.__process_render2d_exception(node)
  43
  44     @staticmethod
  45     def __process_render2d_exception(self, node):
  46         render2d.ls()
  47         message = f'unexpected render2d node: {node.name}'
  48         raise Exception(message)
  49
  50     @staticmethod
  51     def assert_events():
  52         preserve = ['window-event', 'window-closed', 'async_loader_0', 'async_loader_1',
  53                     'render-texture-targets-changed', 'aspectRatioChanged', 'new_scene']
  54         preserve_re = ['^async_loader_.*', '^click-mouse1-pg.*']
  55         unexpected_events = [e for e in messenger.getEvents()
  56                              if e not in preserve and not any(search(r, e) for r in preserve_re)]
  57         for e in unexpected_events: Assert.__process_event_exception(e)
  58
  59     @staticmethod
  60     def __process_event_exception(event):
  61         if (acc := messenger.who_accepts(event)):
  62             Assert.__process_event_acceptors(acc, event)
  63
  64     @staticmethod
  65     def __process_event_acceptors(acceptors, event):
  66         for key, a in acceptors.items(): Assert.__process_event_acceptor(key, a, event)
  67
  68     @staticmethod
  69     def __process_event_acceptor(key, acceptor, event):
  70         #if acceptor[2]:
  71         #    print(f'the event {event} accepted by <{key}, {acceptor[0]}> is persistent')
  72         #else:
  73             Assert.__actually_process_event_exception(acceptor, event)
  74
  75     @staticmethod
  76     def __actually_process_event_exception(acceptor, event):
  77         message = f'unexpected event: {event}, {str(acceptor)}'
  78         raise Exception(message)
  79
  80     @staticmethod
  81     def assert_tasks():
  82         preserve = [
  83             'ivalLoop', 'garbageCollectStates', 'collisionLoop',
  84             'igLoop', 'audioLoop', 'resetPrevTransform', 'dataLoop',
  85             'eventManager', 'simplepbr update', 'on frame music',
  86             'assert_fps', 'DIRECTContextTask']
  87         unexpected_tasks = [t for t in taskMgr.getTasks() + taskMgr.getDoLaters()
  88                             if t.name not in preserve and not hasattr(t, 'preserve')]
  89         for t in unexpected_tasks: Assert.__process_task_exception(t)
  90
  91     @staticmethod
  92     def __process_task_exception(task):
  93         message = f'unexpected task: {task.name}'
  94         raise Exception(message)
  95
  96     @staticmethod
  97     def assert_buffers():
  98         pass
  99         #if RenderToTexture.buffers:
 100         #    raise Error()
 101
 102     @staticmethod
 103     def assert_threads():
 104         thread_names = [thread.name for thread in threading.enumerate()]
 105         preserve = ['MainThread', 'rpc_server']
 106         unexpected_tasks = [t for t in thread_names if t not in preserve]
 107         for t in unexpected_tasks: Assert.__process_unexpected_thread(t)
 108
 109     @staticmethod
 110     def __process_unexpected_thread(thread_name):
 111         message = f'unexpected thread: {thread_name}'
 112         raise Exception(message)