os.environ['PREPAREONLY'] = '1'
      manager.main()
      t = time.monotonic() - start
      assert t < MAX_STARTUP_TIME, f"startup took {t}s, expected <{MAX_STARTUP_TIME}s"

  # ensure all processes exit cleanly
  def test_clean_exit(self):
    manager.manager_prepare()


            

Reported by Bandit.

                    exit_codes = [0, 1]
      if managed_processes[p].sigkill:
        exit_codes = [-signal.SIGKILL]
      assert exit_code in exit_codes, f"{p} died with {exit_code}"


if __name__ == "__main__":
  unittest.main()

            

Reported by Bandit.

              import sys
import matplotlib.pyplot as plt
import numpy as np
import cereal.messaging as messaging
import time

# tool to plot one or more signals live. Call ex:
#./rqplot.py log.carState.vEgo log.carState.aEgo


            

Reported by Pylint.

              import sys
import matplotlib.pyplot as plt
import numpy as np
import cereal.messaging as messaging
import time

# tool to plot one or more signals live. Call ex:
#./rqplot.py log.carState.vEgo log.carState.aEgo


            

Reported by Pylint.

              from typing import List

import numpy as np
import sympy as sp

from rednose.helpers.ekf_sym import EKF_sym, gen_code
from selfdrive.locationd.models.constants import ObservationKind
from selfdrive.locationd.models.loc_kf import parse_pr, parse_prr


            

Reported by Pylint.

              import numpy as np
import sympy as sp

from rednose.helpers.ekf_sym import EKF_sym, gen_code
from selfdrive.locationd.models.constants import ObservationKind
from selfdrive.locationd.models.loc_kf import parse_pr, parse_prr


class States():

            

Reported by Pylint.

              from selfdrive.swaglog import cloudlog
from selfdrive.version import version

import sentry_sdk
from sentry_sdk.integrations.threading import ThreadingIntegration

def capture_exception(*args, **kwargs) -> None:
  cloudlog.error("crash", exc_info=kwargs.get('exc_info', 1))


            

Reported by Pylint.

              from selfdrive.version import version

import sentry_sdk
from sentry_sdk.integrations.threading import ThreadingIntegration

def capture_exception(*args, **kwargs) -> None:
  cloudlog.error("crash", exc_info=kwargs.get('exc_info', 1))

  try:

            

Reported by Pylint.

              import sys
from collections import defaultdict

import cereal.messaging as messaging
from common.realtime import sec_since_boot


def can_printer(bus=0):
  """Collects messages and prints when a new bit transition is observed.

            

Reported by Pylint.

              import sys
from collections import defaultdict

import cereal.messaging as messaging
from common.realtime import sec_since_boot


def can_printer(bus=0):
  """Collects messages and prints when a new bit transition is observed.

            

Reported by Pylint.

              import sys
from collections import defaultdict

import cereal.messaging as messaging
from common.realtime import sec_since_boot


def can_printer(bus=0, max_msg=None, addr="127.0.0.1"):
  logcan = messaging.sub_sock('can', addr=addr)

            

Reported by Pylint.

              import sys
from collections import defaultdict

import cereal.messaging as messaging
from common.realtime import sec_since_boot


def can_printer(bus=0, max_msg=None, addr="127.0.0.1"):
  logcan = messaging.sub_sock('can', addr=addr)

            

Reported by Pylint.

              import numpy as np
from collections import defaultdict, deque
from common.realtime import sec_since_boot
import cereal.messaging as messaging


if __name__ == "__main__":
  context = messaging.Context()
  poller = messaging.Poller()

            

Reported by Pylint.

              import numpy as np
from collections import defaultdict, deque
from common.realtime import sec_since_boot
import cereal.messaging as messaging


if __name__ == "__main__":
  context = messaging.Context()
  poller = messaging.Poller()

            

Reported by Pylint.

                  panda_time = datetime.datetime(a[0], a[1], a[2], a[4], a[5], a[6])
    if panda_time > MIN_DATE:
      logger.info(f"adjusting time from '{sys_time}' to '{panda_time}'")
      os.system(f"TZ=UTC date -s '{panda_time}'")
  except Exception:
    logger.warn("Failed to fetch time from panda")

if __name__ == "__main__":
  import logging

            

Reported by Bandit.

              import datetime
import os
import struct
import usb1

REQUEST_IN = usb1.ENDPOINT_IN | usb1.TYPE_VENDOR | usb1.RECIPIENT_DEVICE
MIN_DATE = datetime.datetime(year=2021, month=4, day=1)

def set_time(logger):

            

Reported by Pylint.

              import numpy as np
from collections import defaultdict, deque

import cereal.messaging as messaging

socks = {s: messaging.sub_sock(s, conflate=False) for s in sys.argv[1:]}
ts = defaultdict(lambda: deque(maxlen=100))

if __name__ == "__main__":

            

Reported by Pylint.

              import numpy as np
from collections import defaultdict, deque

import cereal.messaging as messaging

socks = {s: messaging.sub_sock(s, conflate=False) for s in sys.argv[1:]}
ts = defaultdict(lambda: deque(maxlen=100))

if __name__ == "__main__":

            

Reported by Pylint.

              
    {# compute number of input ports #}
    {%- set n_inputs = 1 %}  {# x0 #}
    {%- if dims.nu > 0 %}  {# u0 -#}
        {%- set n_inputs = n_inputs + 1 -%}
    {%- endif %}
    {%- if dims.np > 0 %}  {# parameters #}
        {%- set n_inputs = n_inputs + 1 -%}
    {%- endif %}

            

Reported by Cppcheck.

              
    {# compute number of input ports #}
    {%- set n_inputs = 1 %}  {# x0 #}
    {%- if dims.nu > 0 %}  {# u0 -#}
        {%- set n_inputs = n_inputs + 1 -%}
    {%- endif %}
    {%- if dims.np > 0 %}  {# parameters #}
        {%- set n_inputs = n_inputs + 1 -%}
    {%- endif %}

            

Reported by Cppcheck.