6380911868

Committed 02 Oct 2023 01:42PM UTC coverage: 33.617% (-0.3%) from 33.898%

Build # 6380911868

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #47 from SpiNNakerManchester/raise_skiptest

Raise skiptest

Run Details

3 of 3 new or added lines in 1 file covered. (100.0%)

79 of 235 relevant lines covered (33.62%)

0.34 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

36.84

/spinnaker_testbase/root_test_case.py

# Copyright (c) 2017 The University of Manchester
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import sys
import time
import unittest
from spinn_utilities.exceptions import NotSetupException
from spinnman.exceptions import SpinnmanException
from pacman.exceptions import PacmanPartitionException, PacmanValueError
from spalloc_client.job import JobDestroyedError
from spinn_front_end_common.data import FecDataView

if os.environ.get('CONTINUOUS_INTEGRATION', 'false').lower() == 'true':
    max_tries = 3
else:
    max_tries = 1


class RootTestCase(unittest.TestCase):

    def _setUp(self, script):
        # Remove random effect for testing
        # Set test_seed to None to allow random
        # pylint: disable=attribute-defined-outside-init
        self._test_seed = 1

        path = os.path.dirname(script)
        os.chdir(path)

    @staticmethod
    def assert_not_spin_three():
        """
        Will raise a SkipTest if run on a none virtual 4 chip board

        :raises SkipTest: If we're on the wrong sort of board
        """
        version = FecDataView.get_machine_version().number
        if not version == 5:
            raise unittest.SkipTest(
                f"This test will not run on a spinn-{version} board")

    def error_file(self):
        """
        The file any error where reported to before a second run attempt

        :return: Path to (possibly non existent) error file
        """
        test_base_directory = os.path.dirname(__file__)
        test_dir = os.path.dirname(test_base_directory)
        return os.path.join(test_dir, "ErrorFile.txt")

    def report(self, message, file_name):
        if not message.endswith("\n"):
            message += "\n"
        global_reports = os.environ.get("GLOBAL_REPORTS", None)
        if not global_reports:
            try:
                global_reports = FecDataView.get_timestamp_dir_path()
            except NotSetupException:
                # This may happen if you are running none script fiels locally
                return

        if not os.path.exists(global_reports):
            # It might now exist if run in parallel
            try:
                os.makedirs(global_reports)
            except Exception:  # pylint: disable=broad-except
                pass
        report_path = os.path.join(global_reports, file_name)
        with open(report_path, "a", encoding="utf-8") as report_file:
            report_file.write(message)

    def runsafe(self, method, retry_delay=3.0, skip_exceptions=None):
        """
        Will run the method possibly a few times

        :param callable method:
        :param float retry_delay:
        :param skip_exceptions:
            list of exception classes to convert in SkipTest
        :type skip_exceptions: list(class)
        """
        if skip_exceptions is None:
            skip_exceptions = []
        retries = 0
        while True:
            try:
                method()
                break
            except (JobDestroyedError, SpinnmanException) as ex:
                for skip_exception in skip_exceptions:
                    if isinstance(ex, skip_exception):
                        FecDataView.raise_skiptest(
                            f"{ex} Still not fixed!", ex)
                class_file = sys.modules[self.__module__].__file__
                with open(self.error_file(), "a", encoding="utf-8") \
                        as error_file:
                    error_file.write(class_file)
                    error_file.write("\n")
                    error_file.write(str(ex))
                    error_file.write("\n")
                retries += 1
                if retries >= max_tries:
                    raise ex
            except (PacmanValueError, PacmanPartitionException) as ex:
                # skip out if on a spin three
                self.assert_not_spin_three()
                for skip_exception in skip_exceptions:
                    if isinstance(ex, skip_exception):
                        FecDataView.raise_skiptest(
                            f"{ex} Still not fixed!", ex)
                raise ex
            print("")
            print("==========================================================")
            print(f" Will run {method} again in {retry_delay} seconds")
            print(f" retry: {retries}")
            print("==========================================================")
            print("")
            time.sleep(retry_delay)

1	# Copyright (c) 2017 The University of Manchester
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# https://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14
15	import os	1✔
16	import sys	1✔
17	import time	1✔
18	import unittest	1✔
19	from spinn_utilities.exceptions import NotSetupException	1✔
20	from spinnman.exceptions import SpinnmanException	1✔
21	from pacman.exceptions import PacmanPartitionException, PacmanValueError	1✔
22	from spalloc_client.job import JobDestroyedError	1✔
23	from spinn_front_end_common.data import FecDataView	1✔
24
25	if os.environ.get('CONTINUOUS_INTEGRATION', 'false').lower() == 'true':	1✔
26	max_tries = 3	×
27	else:
28	max_tries = 1	1✔
29
30
31	class RootTestCase(unittest.TestCase):	1✔
32
33	def _setUp(self, script):	1✔
34	# Remove random effect for testing
35	# Set test_seed to None to allow random
36	# pylint: disable=attribute-defined-outside-init
37	self._test_seed = 1	×
38
39	path = os.path.dirname(script)	×
40	os.chdir(path)	×
41
42	@staticmethod	1✔
43	def assert_not_spin_three():	1✔
44	"""
45	Will raise a SkipTest if run on a none virtual 4 chip board
46
47	:raises SkipTest: If we're on the wrong sort of board
48	"""
49	version = FecDataView.get_machine_version().number	×
50	if not version == 5:	×
51	raise unittest.SkipTest(	×
52	f"This test will not run on a spinn-{version} board")
53
54	def error_file(self):	1✔
55	"""
56	The file any error where reported to before a second run attempt
57
58	:return: Path to (possibly non existent) error file
59	"""
60	test_base_directory = os.path.dirname(__file__)	×
61	test_dir = os.path.dirname(test_base_directory)	×
62	return os.path.join(test_dir, "ErrorFile.txt")	×
63
64	def report(self, message, file_name):	1✔
65	if not message.endswith("\n"):	1✔
66	message += "\n"	1✔
67	global_reports = os.environ.get("GLOBAL_REPORTS", None)	1✔
68	if not global_reports:	1✔
69	try:	1✔
70	global_reports = FecDataView.get_timestamp_dir_path()	1✔
71	except NotSetupException:	×
72	# This may happen if you are running none script fiels locally
73	return	×
74
75	if not os.path.exists(global_reports):	1✔
76	# It might now exist if run in parallel
77	try:	×
78	os.makedirs(global_reports)	×
79	except Exception: # pylint: disable=broad-except	×
80	pass	×
81	report_path = os.path.join(global_reports, file_name)	1✔
82	with open(report_path, "a", encoding="utf-8") as report_file:	1✔
83	report_file.write(message)	1✔
84
85	def runsafe(self, method, retry_delay=3.0, skip_exceptions=None):	1✔
86	"""
87	Will run the method possibly a few times
88
89	:param callable method:
90	:param float retry_delay:
91	:param skip_exceptions:
92	list of exception classes to convert in SkipTest
93	:type skip_exceptions: list(class)
94	"""
95	if skip_exceptions is None:	×
96	skip_exceptions = []	×
97	retries = 0	×
98	while True:
99	try:	×
100	method()	×
101	break	×
102	except (JobDestroyedError, SpinnmanException) as ex:	×
103	for skip_exception in skip_exceptions:	×
104	if isinstance(ex, skip_exception):	×
105	FecDataView.raise_skiptest(	×
106	f"{ex} Still not fixed!", ex)
107	class_file = sys.modules[self.__module__].__file__	×
108	with open(self.error_file(), "a", encoding="utf-8") \	×
109	as error_file:
110	error_file.write(class_file)	×
111	error_file.write("\n")	×
112	error_file.write(str(ex))	×
113	error_file.write("\n")	×
114	retries += 1	×
115	if retries >= max_tries:	×
116	raise ex	×
117	except (PacmanValueError, PacmanPartitionException) as ex:	×
118	# skip out if on a spin three
119	self.assert_not_spin_three()	×
120	for skip_exception in skip_exceptions:	×
121	if isinstance(ex, skip_exception):	×
122	FecDataView.raise_skiptest(	×
123	f"{ex} Still not fixed!", ex)
124	raise ex	×
125	print("")	×
126	print("==========================================================")	×
127	print(f" Will run {method} again in {retry_delay} seconds")	×
128	print(f" retry: {retries}")	×
129	print("==========================================================")	×
130	print("")	×
131	time.sleep(retry_delay)	×

SpiNNakerManchester / TestBase / 6380911868

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous