diff --git a/core/opl/data.py b/core/opl/data.py index 8a6e4dd..d4d1e27 100644 --- a/core/opl/data.py +++ b/core/opl/data.py @@ -196,7 +196,9 @@ def data_stats(data): data = [i for i in data if isinstance(i, datetime.datetime) or math.isfinite(i)] count_strange = data_len_before - len(data) if count_strange > 0: - logging.warning(f"There were {count_strange} NaN/Inf values in the data. Filtered them out.") + logging.warning( + f"There were {count_strange} NaN/Inf values in the data. Filtered them out." + ) non_zero_data = [i for i in data if i != 0] diff --git a/core/opl/status_data_updater.py b/core/opl/status_data_updater.py index 8d41bc0..ec84f57 100755 --- a/core/opl/status_data_updater.py +++ b/core/opl/status_data_updater.py @@ -314,7 +314,11 @@ def _get_es_result_for_rp_result(session, args, run_id, result): def _get_es_dashboard_result_for_run_id(session, args, run_id, test=None): if test is not None: response = _es_get_test( - session, args, ["result_id.keyword", "test.keyword"], [run_id, test], sort_by="date" + session, + args, + ["result_id.keyword", "test.keyword"], + [run_id, test], + sort_by="date", ) else: response = _es_get_test( @@ -466,7 +470,9 @@ def doit_rp_to_dashboard_new(args): session, args, run_id ) except requests.exceptions.HTTPError as e: - matching = "No mapping found for [date] in order to sort on" in e.response.text + matching = ( + "No mapping found for [date] in order to sort on" in e.response.text + ) if e.response.status_code == 400 and matching: logging.debug( "Request failed, but I guess it was because index is still empty" @@ -503,6 +509,7 @@ def doit_rp_to_dashboard_new(args): ) print(f"Created result {run_id} in the dashboard with value {result}") + def _update_es_dashboard_result(session, args, es_id, result_string): url = f"{args.es_server}/{args.es_index}/_doc/{es_id}/_update" headers = { @@ -526,6 +533,7 @@ def _update_es_dashboard_result(session, args, es_id, result_string): f"Got back this: {json.dumps(response.json(), sort_keys=True, indent=4)}" ) + def doit_rp_to_dashboard_update(args): assert args.es_server is not None assert args.rp_host is not None @@ -566,7 +574,10 @@ def doit_rp_to_dashboard_update(args): # Get relevant dashboard result from ElasticSearch dashboard, es_type, es_id = _get_es_dashboard_result_for_run_id( - session, args, run_id, result["name"], + session, + args, + run_id, + result["name"], ) if dashboard is None: logging.warning( @@ -580,7 +591,10 @@ def doit_rp_to_dashboard_update(args): pass # data in the dashboard are correct, no action needed else: _update_es_dashboard_result( - session, args, es_id, result_string, + session, + args, + es_id, + result_string, ) stats["results_changed"] += 1 diff --git a/core/setup.py b/core/setup.py index 0af0362..a4fcac3 100644 --- a/core/setup.py +++ b/core/setup.py @@ -22,7 +22,7 @@ "Intended Audience :: Developers", "Topic :: Software Development :: Quality Assurance", ], - python_requires='>=3.6', + python_requires=">=3.6", install_requires=[ "Jinja2>=3.0", "boto3", diff --git a/opl/data.py b/opl/data.py index 8a6e4dd..d4d1e27 100644 --- a/opl/data.py +++ b/opl/data.py @@ -196,7 +196,9 @@ def data_stats(data): data = [i for i in data if isinstance(i, datetime.datetime) or math.isfinite(i)] count_strange = data_len_before - len(data) if count_strange > 0: - logging.warning(f"There were {count_strange} NaN/Inf values in the data. Filtered them out.") + logging.warning( + f"There were {count_strange} NaN/Inf values in the data. Filtered them out." + ) non_zero_data = [i for i in data if i != 0] diff --git a/opl/status_data_updater.py b/opl/status_data_updater.py index 8d41bc0..ec84f57 100755 --- a/opl/status_data_updater.py +++ b/opl/status_data_updater.py @@ -314,7 +314,11 @@ def _get_es_result_for_rp_result(session, args, run_id, result): def _get_es_dashboard_result_for_run_id(session, args, run_id, test=None): if test is not None: response = _es_get_test( - session, args, ["result_id.keyword", "test.keyword"], [run_id, test], sort_by="date" + session, + args, + ["result_id.keyword", "test.keyword"], + [run_id, test], + sort_by="date", ) else: response = _es_get_test( @@ -466,7 +470,9 @@ def doit_rp_to_dashboard_new(args): session, args, run_id ) except requests.exceptions.HTTPError as e: - matching = "No mapping found for [date] in order to sort on" in e.response.text + matching = ( + "No mapping found for [date] in order to sort on" in e.response.text + ) if e.response.status_code == 400 and matching: logging.debug( "Request failed, but I guess it was because index is still empty" @@ -503,6 +509,7 @@ def doit_rp_to_dashboard_new(args): ) print(f"Created result {run_id} in the dashboard with value {result}") + def _update_es_dashboard_result(session, args, es_id, result_string): url = f"{args.es_server}/{args.es_index}/_doc/{es_id}/_update" headers = { @@ -526,6 +533,7 @@ def _update_es_dashboard_result(session, args, es_id, result_string): f"Got back this: {json.dumps(response.json(), sort_keys=True, indent=4)}" ) + def doit_rp_to_dashboard_update(args): assert args.es_server is not None assert args.rp_host is not None @@ -566,7 +574,10 @@ def doit_rp_to_dashboard_update(args): # Get relevant dashboard result from ElasticSearch dashboard, es_type, es_id = _get_es_dashboard_result_for_run_id( - session, args, run_id, result["name"], + session, + args, + run_id, + result["name"], ) if dashboard is None: logging.warning( @@ -580,7 +591,10 @@ def doit_rp_to_dashboard_update(args): pass # data in the dashboard are correct, no action needed else: _update_es_dashboard_result( - session, args, es_id, result_string, + session, + args, + es_id, + result_string, ) stats["results_changed"] += 1 diff --git a/setup.py b/setup.py index 0a335e6..620bd96 100644 --- a/setup.py +++ b/setup.py @@ -22,7 +22,7 @@ "Intended Audience :: Developers", "Topic :: Software Development :: Quality Assurance", ], - python_requires='>=3.6', + python_requires=">=3.6", install_requires=[ "Jinja2>=3.0", "boto3", diff --git a/tests/context.py b/tests/context.py index 24efd3f..19614b5 100644 --- a/tests/context.py +++ b/tests/context.py @@ -2,14 +2,15 @@ import os import sys -sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) -import opl.generators.qpc_tarball # noqa: E402 -import opl.generators.inventory_ingress # noqa: E402 -import opl.generators.inventory_egress # noqa: E402 -import opl.generators.packages # noqa: E402 -import opl.status_data # noqa: E402 -import opl.cluster_read # noqa: E402 -import opl.junit_cli # noqa: E402 -import opl.args # noqa: E402 -import opl # noqa: E402 F401 +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) + +import opl.generators.qpc_tarball # noqa: E402 +import opl.generators.inventory_ingress # noqa: E402 +import opl.generators.inventory_egress # noqa: E402 +import opl.generators.packages # noqa: E402 +import opl.status_data # noqa: E402 +import opl.cluster_read # noqa: E402 +import opl.junit_cli # noqa: E402 +import opl.args # noqa: E402 +import opl # noqa: E402 F401 diff --git a/tests/test_args.py b/tests/test_args.py index 21c570f..cd7434f 100644 --- a/tests/test_args.py +++ b/tests/test_args.py @@ -7,8 +7,7 @@ class TestArgs(unittest.TestCase): - def test_add_storage_db_opts(self): parser = argparse.ArgumentParser() opl.args.add_storage_db_opts(parser) - self.assertIn('--storage-db-host', parser.format_help()) + self.assertIn("--storage-db-host", parser.format_help()) diff --git a/tests/test_cluster_read.py b/tests/test_cluster_read.py index f20bed2..b393680 100755 --- a/tests/test_cluster_read.py +++ b/tests/test_cluster_read.py @@ -25,7 +25,7 @@ def test_date(self): before = datetime.datetime.utcnow().year k, v = next(ri) after = datetime.datetime.utcnow().year - self.assertEqual(k, 'mydate') + self.assertEqual(k, "mydate") self.assertGreaterEqual(int(v), before) self.assertGreaterEqual(after, int(v)) @@ -37,9 +37,9 @@ def test_json(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'myjson') - self.assertEqual(v['aaa'], 123) - self.assertEqual(v['bbb'], 456) + self.assertEqual(k, "myjson") + self.assertEqual(v["aaa"], 123) + self.assertEqual(v["bbb"], 456) def test_yaml(self): string = """ @@ -49,14 +49,14 @@ def test_yaml(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'myyaml') - self.assertEqual(v['aaa'], 123) - self.assertEqual(v['bbb'], 456) + self.assertEqual(k, "myyaml") + self.assertEqual(v["aaa"], 123) + self.assertEqual(v["bbb"], 456) def test_measurements(self): class TestMeasurementPlugin(opl.cluster_read.BasePlugin): def measure(self, ri, name, test_measurement_query): - if test_measurement_query == 'simple': + if test_measurement_query == "simple": return name, opl.data.data_stats([1, 2, 3]) string = """ @@ -64,11 +64,13 @@ def measure(self, ri, name, test_measurement_query): test_measurement_query: simple """ ri = opl.cluster_read.RequestedInfo(string) - ri.register_measurement_plugin('test_measurement_query', TestMeasurementPlugin({})) + ri.register_measurement_plugin( + "test_measurement_query", TestMeasurementPlugin({}) + ) k, v = next(ri) - self.assertEqual(k, 'mymeasurement') - self.assertEqual(v['samples'], 3) - self.assertEqual(v['mean'], 2) + self.assertEqual(k, "mymeasurement") + self.assertEqual(v["samples"], 3) + self.assertEqual(v["mean"], 2) def test_config_type(self): string = """ @@ -77,16 +79,16 @@ def test_config_type(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'mygreet') - self.assertEqual(v, 'hello') + self.assertEqual(k, "mygreet") + self.assertEqual(v, "hello") tmp_file = tempfile.mkstemp()[1] - with open(tmp_file, 'w') as fpw: + with open(tmp_file, "w") as fpw: fpw.write(string) - with open(tmp_file, 'r') as fpr: + with open(tmp_file, "r") as fpr: ri = opl.cluster_read.RequestedInfo(fpr) k, v = next(ri) - self.assertEqual(k, 'mygreet') - self.assertEqual(v, 'hello') + self.assertEqual(k, "mygreet") + self.assertEqual(v, "hello") os.remove(tmp_file) def test_jinja2_config(self): @@ -96,14 +98,14 @@ def test_jinja2_config(self): command: echo '{{ SOMETHING }}-{{ item }}' {% endfor %} """ - os.environ['SOMETHING'] = 'foobarbaz' + os.environ["SOMETHING"] = "foobarbaz" ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'myenv-1') - self.assertEqual(v, 'foobarbaz-1') + self.assertEqual(k, "myenv-1") + self.assertEqual(v, "foobarbaz-1") k, v = next(ri) - self.assertEqual(k, 'myenv-2') - self.assertEqual(v, 'foobarbaz-2') + self.assertEqual(k, "myenv-2") + self.assertEqual(v, "foobarbaz-2") def test_get_config(self): string = """ @@ -121,8 +123,8 @@ def test_constant(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'myconstant') - self.assertEqual(v, 'Hello world') + self.assertEqual(k, "myconstant") + self.assertEqual(v, "Hello world") def test_copy_from(self): string = """ @@ -133,11 +135,11 @@ def test_copy_from(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'somevalue') - self.assertEqual(v, 'Hello world') + self.assertEqual(k, "somevalue") + self.assertEqual(v, "Hello world") k, v = next(ri) - self.assertEqual(k, 'mycopyfrom') - self.assertEqual(v, 'Hello world') + self.assertEqual(k, "mycopyfrom") + self.assertEqual(v, "Hello world") def test_copy_from_negative(self): string = """ @@ -148,10 +150,10 @@ def test_copy_from_negative(self): """ ri = opl.cluster_read.RequestedInfo(string) k, v = next(ri) - self.assertEqual(k, 'somevalue') - self.assertEqual(v, 'Hello world') + self.assertEqual(k, "somevalue") + self.assertEqual(v, "Hello world") k, v = next(ri) - self.assertEqual(k, 'mycopyfrom') + self.assertEqual(k, "mycopyfrom") self.assertEqual(v, None) def test_wrong_config(self): diff --git a/tests/test_generatore_insights_ingress.py b/tests/test_generatore_insights_ingress.py index 6a97327..0a07923 100644 --- a/tests/test_generatore_insights_ingress.py +++ b/tests/test_generatore_insights_ingress.py @@ -26,10 +26,10 @@ def test_count(self): def test_templates(self): pg = opl.generators.inventory_ingress.PayloadRHSMGenerator( - count=1, - template='inventory_ingress_RHSM_template.json.j2') + count=1, template="inventory_ingress_RHSM_template.json.j2" + ) mid, msg = next(pg) pg = opl.generators.inventory_ingress.PayloadRHSMGenerator( - count=1, - template='inventory_ingress_yupana_template.json.j2') + count=1, template="inventory_ingress_yupana_template.json.j2" + ) mid, msg = next(pg) diff --git a/tests/test_generators_insights_egress.py b/tests/test_generators_insights_egress.py index 97e438b..d60f83e 100644 --- a/tests/test_generators_insights_egress.py +++ b/tests/test_generators_insights_egress.py @@ -14,12 +14,11 @@ class TestPayloadGenerator(unittest.TestCase): def test_counting(self): pg = opl.generators.inventory_egress.EgressHostsGenerator( - count=3, - n_packages=10, - msg_type='created') + count=3, n_packages=10, msg_type="created" + ) self.assertEqual(pg.count, 3) self.assertEqual(pg.n_packages, 10) - self.assertEqual(pg.msg_type, 'created') + self.assertEqual(pg.msg_type, "created") def test_count(self): pg = opl.generators.inventory_egress.EgressHostsGenerator() diff --git a/tests/test_generators_packages.py b/tests/test_generators_packages.py index c70c45f..4539d93 100644 --- a/tests/test_generators_packages.py +++ b/tests/test_generators_packages.py @@ -25,6 +25,8 @@ def test_looks_like_package_name(self): pg = opl.generators.packages.PackagesGenerator() for p in pg.generate(10): self.assertIsInstance(p, str) - self.assertTrue(p.endswith('x86_64') or p.endswith('i686') or p.endswith('noarch')) - self.assertTrue(' ' not in p) - self.assertGreater(len(p), 9) # minimum is 'a-1.noarch' I think + self.assertTrue( + p.endswith("x86_64") or p.endswith("i686") or p.endswith("noarch") + ) + self.assertTrue(" " not in p) + self.assertGreater(len(p), 9) # minimum is 'a-1.noarch' I think diff --git a/tests/test_generators_qpc_tarball.py b/tests/test_generators_qpc_tarball.py index d0eabcc..7878326 100644 --- a/tests/test_generators_qpc_tarball.py +++ b/tests/test_generators_qpc_tarball.py @@ -60,7 +60,6 @@ def test_add_hosts(self): tarball_slice.add_host({"fqdn": f"host{i}.example.com"}) self.assertEqual(tarball_slice.get_host_count(), 4) - def test_slice_after_dump(self): pg = opl.generators.qpc_tarball.QPCTarballGenerator( count=2, tarball_conf=self.tarball_conf diff --git a/tests/test_junit_cli.py b/tests/test_junit_cli.py index 7532a96..b7501ca 100755 --- a/tests/test_junit_cli.py +++ b/tests/test_junit_cli.py @@ -34,29 +34,29 @@ def test_fromfile_or_new(self): def test_add_to_suite(self): tc1 = { - 'name': 'bbb', - 'result': 'PASS', - 'message': None, - 'system-out': None, - 'system-err': None, - 'start': opl.date.my_fromisoformat('2019-12-18T14:05:33+01:00'), - 'end': opl.date.my_fromisoformat('2019-12-18T14:10:56+01:00'), + "name": "bbb", + "result": "PASS", + "message": None, + "system-out": None, + "system-err": None, + "start": opl.date.my_fromisoformat("2019-12-18T14:05:33+01:00"), + "end": opl.date.my_fromisoformat("2019-12-18T14:10:56+01:00"), } tc2 = tc1.copy() - tc2['name'] = 'ccc' - tc2['result'] = 'FAIL' + tc2["name"] = "ccc" + tc2["result"] = "FAIL" - self.junit.add_to_suite('aaa', tc1) - self.junit.add_to_suite('aaa', tc2) + self.junit.add_to_suite("aaa", tc1) + self.junit.add_to_suite("aaa", tc2) suite = next(iter(self.junit)) - self.assertEqual(suite.name, 'aaa') + self.assertEqual(suite.name, "aaa") suite_iter = iter(suite) case1 = next(suite_iter) case2 = next(suite_iter) - self.assertEqual(case1.name, 'bbb') + self.assertEqual(case1.name, "bbb") self.assertEqual(case1.result, []) - self.assertEqual(case2.name, 'ccc') + self.assertEqual(case2.name, "ccc") self.assertEqual(len(case2.result), 1) self.assertEqual(type(case2.result[0]), junitparser.junitparser.Failure) diff --git a/tests/test_skelet.py b/tests/test_skelet.py index 1bd187c..1e4cb10 100644 --- a/tests/test_skelet.py +++ b/tests/test_skelet.py @@ -7,9 +7,10 @@ class TestSkelet(unittest.TestCase): - def test_test_setup(self): parser = argparse.ArgumentParser() - parser.add_argument('--fake_arg', help="Just to catch test file name when calling tests") + parser.add_argument( + "--fake_arg", help="Just to catch test file name when calling tests" + ) with opl.skelet.test_setup(parser) as (args, status_data): - self.assertIn('debug', args) + self.assertIn("debug", args) diff --git a/tests/test_status_data.py b/tests/test_status_data.py index a14e563..fa8f3e1 100755 --- a/tests/test_status_data.py +++ b/tests/test_status_data.py @@ -22,224 +22,239 @@ def setUp(self): self.status_data = opl.status_data.StatusData(self.tmpfile) def tierDown(self): - del(self.status_data) + del self.status_data os.remove(self.tmpfile) def test_filename(self): self.assertEqual(self.status_data._filename, self.tmpfile) def test_defaults(self): - self.assertIn('name', self.status_data._data) - self.assertIn('owner', self.status_data._data) - self.assertIn('started', self.status_data._data) - self.assertIn('ended', self.status_data._data) - self.assertIn('result', self.status_data._data) + self.assertIn("name", self.status_data._data) + self.assertIn("owner", self.status_data._data) + self.assertIn("started", self.status_data._data) + self.assertIn("ended", self.status_data._data) + self.assertIn("result", self.status_data._data) def test_simple_set(self): - self.status_data.set('aaa', 123) - self.assertEqual(self.status_data._data['aaa'], 123) + self.status_data.set("aaa", 123) + self.assertEqual(self.status_data._data["aaa"], 123) def test_nested_set(self): - self.status_data.set('aaa.bbb', 123) - self.assertEqual(self.status_data._data['aaa']['bbb'], 123) + self.status_data.set("aaa.bbb", 123) + self.assertEqual(self.status_data._data["aaa"]["bbb"], 123) def test_simple_update(self): - self.status_data.set('aaa', 123) - self.status_data.set('aaa', 456) - self.assertEqual(self.status_data._data['aaa'], 456) + self.status_data.set("aaa", 123) + self.status_data.set("aaa", 456) + self.assertEqual(self.status_data._data["aaa"], 456) def test_nested_update(self): - self.status_data.set('aaa.bbb', 123) - self.status_data.set('aaa.bbb', 456) - self.assertEqual(self.status_data._data['aaa']['bbb'], 456) + self.status_data.set("aaa.bbb", 123) + self.status_data.set("aaa.bbb", 456) + self.assertEqual(self.status_data._data["aaa"]["bbb"], 456) def test_additional_set(self): - self.status_data.set('aaa.bbb', 123) - self.assertEqual(self.status_data._data['aaa']['bbb'], 123) - self.status_data.set('aaa.ccc', 456) - self.assertEqual(self.status_data._data['aaa']['bbb'], 123) - self.assertEqual(self.status_data._data['aaa']['ccc'], 456) + self.status_data.set("aaa.bbb", 123) + self.assertEqual(self.status_data._data["aaa"]["bbb"], 123) + self.status_data.set("aaa.ccc", 456) + self.assertEqual(self.status_data._data["aaa"]["bbb"], 123) + self.assertEqual(self.status_data._data["aaa"]["ccc"], 456) def test_simple_get(self): - self.status_data._data['aaa'] = 123 - self.assertEqual(self.status_data.get('aaa'), 123) + self.status_data._data["aaa"] = 123 + self.assertEqual(self.status_data.get("aaa"), 123) def test_nested_get(self): - self.status_data._data['aaa'] = {} - self.status_data._data['aaa']['bbb'] = 123 - self.assertEqual(self.status_data.get('aaa.bbb'), 123) + self.status_data._data["aaa"] = {} + self.status_data._data["aaa"]["bbb"] = 123 + self.assertEqual(self.status_data.get("aaa.bbb"), 123) def test_none_get(self): - self.assertIsNone(self.status_data.get('aaa')) - self.assertIsNone(self.status_data.get('aaa.bbb')) - self.assertIsNone(self.status_data.get('aaa.bbb.ccc')) + self.assertIsNone(self.status_data.get("aaa")) + self.assertIsNone(self.status_data.get("aaa.bbb")) + self.assertIsNone(self.status_data.get("aaa.bbb.ccc")) def test_datetime(self): - now_plus2 = datetime.datetime.now(tz=datetime.timezone(datetime.timedelta(hours=2))) + now_plus2 = datetime.datetime.now( + tz=datetime.timezone(datetime.timedelta(hours=2)) + ) now_utc = datetime.datetime.now(tz=datetime.timezone.utc) - self.status_data.set('aaa', now_plus2) - self.status_data.set('bbb', now_utc) - self.assertEqual(self.status_data.get('aaa'), now_plus2.isoformat()) - self.assertEqual(self.status_data.get('bbb'), now_utc.isoformat()) - self.assertEqual(self.status_data.get_date('aaa'), now_plus2) - self.assertEqual(self.status_data.get_date('bbb'), now_utc) + self.status_data.set("aaa", now_plus2) + self.status_data.set("bbb", now_utc) + self.assertEqual(self.status_data.get("aaa"), now_plus2.isoformat()) + self.assertEqual(self.status_data.get("bbb"), now_utc.isoformat()) + self.assertEqual(self.status_data.get_date("aaa"), now_plus2) + self.assertEqual(self.status_data.get_date("bbb"), now_utc) def test_datetime(self): - refference = datetime.datetime(2020, 12, 2, hour=12, minute=0, second=0, microsecond=0, tzinfo=datetime.timezone.utc) + refference = datetime.datetime( + 2020, + 12, + 2, + hour=12, + minute=0, + second=0, + microsecond=0, + tzinfo=datetime.timezone.utc, + ) tests = [ - '2020-12-02T12:00:00+00:00', - '2020-12-02T12:00:00Z', - '2020-12-02T13:00:00+01:00', - '2020-12-02T14:00:00+02:00', - '2020-12-02T12:00:00.000+00:00', - '2020-12-02T12:00:00.000Z', - '2020-12-02T13:00:00.000+01:00', - '2020-12-02T14:00:00.000+02:00', - '2020-12-02T12:00:00.000000+00:00', - '2020-12-02T12:00:00.000000Z', - '2020-12-02T13:00:00.000000+01:00', - '2020-12-02T14:00:00.000000+02:00', + "2020-12-02T12:00:00+00:00", + "2020-12-02T12:00:00Z", + "2020-12-02T13:00:00+01:00", + "2020-12-02T14:00:00+02:00", + "2020-12-02T12:00:00.000+00:00", + "2020-12-02T12:00:00.000Z", + "2020-12-02T13:00:00.000+01:00", + "2020-12-02T14:00:00.000+02:00", + "2020-12-02T12:00:00.000000+00:00", + "2020-12-02T12:00:00.000000Z", + "2020-12-02T13:00:00.000000+01:00", + "2020-12-02T14:00:00.000000+02:00", ] for t in tests: - self.status_data.set('aaa', t) - self.assertEqual(self.status_data.get_date('aaa'), refference) + self.status_data.set("aaa", t) + self.assertEqual(self.status_data.get_date("aaa"), refference) def test_set_now(self): before = datetime.datetime.now(tz=datetime.timezone.utc) - self.status_data.set_now('aaa') + self.status_data.set_now("aaa") after = datetime.datetime.now(tz=datetime.timezone.utc) - aaa = self.status_data.get_date('aaa') + aaa = self.status_data.get_date("aaa") self.assertGreaterEqual(aaa, before) self.assertGreaterEqual(after, aaa) def test_list(self): - self.status_data._data['aaa'] = { - 'bbb': 'ccc', - 'ddd': { - 'eee': 'fff', - 'ggg': 42, - } + self.status_data._data["aaa"] = { + "bbb": "ccc", + "ddd": { + "eee": "fff", + "ggg": 42, + }, } - exp = ['aaa.bbb', 'aaa.ddd.eee', 'aaa.ddd.ggg'] - self.assertCountEqual(self.status_data.list('aaa'), exp) + exp = ["aaa.bbb", "aaa.ddd.eee", "aaa.ddd.ggg"] + self.assertCountEqual(self.status_data.list("aaa"), exp) def test_list_results(self): - self.assertCountEqual(self.status_data.list('results'), []) + self.assertCountEqual(self.status_data.list("results"), []) def test_info(self): - self.status_data.set('aaa', 123) - self.status_data.set('bbb.ccc', 456) - self.assertIn('Filename:', self.status_data.info()) - self.assertIn('aaa', self.status_data.info()) - self.assertIn('123', self.status_data.info()) - self.assertNotIn('bbb', self.status_data.info()) - self.assertNotIn('456', self.status_data.info()) + self.status_data.set("aaa", 123) + self.status_data.set("bbb.ccc", 456) + self.assertIn("Filename:", self.status_data.info()) + self.assertIn("aaa", self.status_data.info()) + self.assertIn("123", self.status_data.info()) + self.assertNotIn("bbb", self.status_data.info()) + self.assertNotIn("456", self.status_data.info()) def test_data(self): f = tempfile.NamedTemporaryFile().name data = { - 'aaa': 123, - 'bbb': { - 'ccc': 456, + "aaa": 123, + "bbb": { + "ccc": 456, }, - 'name': '', - 'started': '', - 'ended': '', - 'result': '', + "name": "", + "started": "", + "ended": "", + "result": "", } sd = opl.status_data.StatusData(f, data=data) - self.assertEqual(sd.get('aaa'), 123) - self.assertEqual(sd.get('bbb.ccc'), 456) + self.assertEqual(sd.get("aaa"), 123) + self.assertEqual(sd.get("bbb.ccc"), 456) def test_long_missing_path(self): f = tempfile.NamedTemporaryFile().name data = { - 'aaa': { - 'bbb': {}, - 'ccc': 123, + "aaa": { + "bbb": {}, + "ccc": 123, }, - 'name': '', - 'started': '', - 'ended': '', - 'result': '', + "name": "", + "started": "", + "ended": "", + "result": "", } sd = opl.status_data.StatusData(f, data=data) - self.assertEqual(sd.get('aaa.ccc'), 123) - self.assertEqual(sd.get('aaa.bbb'), {}) - self.assertEqual(sd.get('aaa.bbb.ddd'), None) - self.assertEqual(sd.get('aaa.bbb.ddd.eee'), None) + self.assertEqual(sd.get("aaa.ccc"), 123) + self.assertEqual(sd.get("aaa.bbb"), {}) + self.assertEqual(sd.get("aaa.bbb.ddd"), None) + self.assertEqual(sd.get("aaa.bbb.ddd.eee"), None) def test_missing_path_in_none(self): f = tempfile.NamedTemporaryFile().name data = { - 'aaa': None, - 'name': '', - 'started': '', - 'ended': '', - 'result': '', + "aaa": None, + "name": "", + "started": "", + "ended": "", + "result": "", } sd = opl.status_data.StatusData(f, data=data) - self.assertEqual(sd.get('aaa'), None) - self.assertEqual(sd.get('aaa.bbb'), None) - self.assertEqual(sd.get('aaa.bbb.ccc'), None) + self.assertEqual(sd.get("aaa"), None) + self.assertEqual(sd.get("aaa.bbb"), None) + self.assertEqual(sd.get("aaa.bbb.ccc"), None) def test_copy_original_object(self): something = { - 'foo': 1, - 'bar': 2, + "foo": 1, + "bar": 2, } - self.status_data.set('results.something', something) - something['baz'] = 3 - self.assertEqual(self.status_data.get('results.something.foo'), 1) - self.assertEqual(self.status_data.get('results.something.bar'), 2) - self.assertIn('baz', something) - self.assertEqual(self.status_data.get('results.something.baz'), None) + self.status_data.set("results.something", something) + something["baz"] = 3 + self.assertEqual(self.status_data.get("results.something.foo"), 1) + self.assertEqual(self.status_data.get("results.something.bar"), 2) + self.assertIn("baz", something) + self.assertEqual(self.status_data.get("results.something.baz"), None) def test_set_subtree_json(self): - with tempfile.NamedTemporaryFile(delete=False, suffix='.json') as f: + with tempfile.NamedTemporaryFile(delete=False, suffix=".json") as f: f_name = f.name f.write(b'{"hello":"world","foo":42,"bar":{"baz":1}}') - self.status_data.set_subtree_json('results.xxx', f_name) + self.status_data.set_subtree_json("results.xxx", f_name) os.unlink(f_name) - self.assertEqual(self.status_data.get('results.xxx.hello'), 'world') - self.assertEqual(self.status_data.get('results.xxx.foo'), 42) - self.assertEqual(self.status_data.get('results.xxx.bar.baz'), 1) + self.assertEqual(self.status_data.get("results.xxx.hello"), "world") + self.assertEqual(self.status_data.get("results.xxx.foo"), 42) + self.assertEqual(self.status_data.get("results.xxx.bar.baz"), 1) def test_set_subtree_yaml(self): - with tempfile.NamedTemporaryFile(delete=False, suffix='.yaml') as f: + with tempfile.NamedTemporaryFile(delete=False, suffix=".yaml") as f: f_name = f.name - f.write(b'hello: world\nfoo: 42\nbar:\n baz: 1') - self.status_data.set_subtree_json('results.xxx', f_name) + f.write(b"hello: world\nfoo: 42\nbar:\n baz: 1") + self.status_data.set_subtree_json("results.xxx", f_name) os.unlink(f_name) - self.assertEqual(self.status_data.get('results.xxx.hello'), 'world') - self.assertEqual(self.status_data.get('results.xxx.foo'), 42) - self.assertEqual(self.status_data.get('results.xxx.bar.baz'), 1) + self.assertEqual(self.status_data.get("results.xxx.hello"), "world") + self.assertEqual(self.status_data.get("results.xxx.foo"), 42) + self.assertEqual(self.status_data.get("results.xxx.bar.baz"), 1) def test_remove_simple(self): - self.status_data.set('results.xxx', 'should not be here') - self.assertEqual(self.status_data.get('results.xxx'), 'should not be here') - self.status_data.remove('results.xxx') - self.assertIsNone(self.status_data.get('results.xxx')) + self.status_data.set("results.xxx", "should not be here") + self.assertEqual(self.status_data.get("results.xxx"), "should not be here") + self.status_data.remove("results.xxx") + self.assertIsNone(self.status_data.get("results.xxx")) def test_remove_missing(self): - self.assertIsNone(self.status_data.get('results.missing')) - self.status_data.remove('results.missing') - self.assertIsNone(self.status_data.get('results.missing')) + self.assertIsNone(self.status_data.get("results.missing")) + self.status_data.remove("results.missing") + self.assertIsNone(self.status_data.get("results.missing")) def test_file_on_http(self): with self.assertRaises(requests.exceptions.ConnectionError): - status_data = opl.status_data.StatusData('http://does.not.exist/status-data-file.json') + status_data = opl.status_data.StatusData( + "http://does.not.exist/status-data-file.json" + ) def test_comment(self): comment = { - 'author': 'Foo Bar', - 'date': datetime.datetime.utcnow().replace(tzinfo=datetime.timezone.utc).isoformat(), - 'text': 'Some comment', + "author": "Foo Bar", + "date": datetime.datetime.utcnow() + .replace(tzinfo=datetime.timezone.utc) + .isoformat(), + "text": "Some comment", } - self.status_data.set('comments', []) - comments = self.status_data.get('comments') + self.status_data.set("comments", []) + comments = self.status_data.get("comments") comments.append(comment) - self.assertEqual(self.status_data.get('comments')[0], comment) + self.assertEqual(self.status_data.get("comments")[0], comment) data = self.status_data.dump() - sd_new = opl.status_data.StatusData('/tmp/aaa.json', data=data) - self.assertEqual(sd_new.get('comments')[0], comment) + sd_new = opl.status_data.StatusData("/tmp/aaa.json", data=data) + self.assertEqual(sd_new.get("comments")[0], comment)