diff --git a/openc3/python/openc3/utilities/store.py b/openc3/python/openc3/utilities/store.py index bb0c379159..024faf654f 100644 --- a/openc3/python/openc3/utilities/store.py +++ b/openc3/python/openc3/utilities/store.py @@ -21,7 +21,7 @@ else: openc3_redis_cluster = False -from openc3.utilities.store_implementation import Store, StoreConnectionPool, StoreMeta, EphemeralStore +from openc3.utilities.store_implementation import Store, StoreConnectionPool, StoreMeta, EphemeralStore # noqa: F401 if openc3_redis_cluster: - import openc3enterprise.utilities.store + import openc3enterprise.utilities.store # noqa: F401 diff --git a/openc3/python/test/conversions/test_object_read_conversion.py b/openc3/python/test/conversions/test_object_read_conversion.py index 7d2504a351..e10ef35174 100644 --- a/openc3/python/test/conversions/test_object_read_conversion.py +++ b/openc3/python/test/conversions/test_object_read_conversion.py @@ -34,7 +34,7 @@ def test_takes_cmd_tlm_target_name_packet_name(self): self.assertEqual(orc.converted_bit_size, 0) def test_complains_about_invalid_cmd_tlm(self): - with self.assertRaisesRegex(AttributeError, f"Unknown type:OTHER"): + with self.assertRaisesRegex(AttributeError, "Unknown type:OTHER"): ObjectReadConversion("OTHER", "TGT", "PKT") def test_fills_the_cmd_packet_and_returns_a_hash_of_the_converted_values(self): diff --git a/openc3/python/test/conversions/test_object_write_conversion.py b/openc3/python/test/conversions/test_object_write_conversion.py index aea0a2218d..dec9f9f55c 100644 --- a/openc3/python/test/conversions/test_object_write_conversion.py +++ b/openc3/python/test/conversions/test_object_write_conversion.py @@ -34,7 +34,7 @@ def test_takes_cmd_tlm_target_name_packet_name(self): self.assertEqual(owc.converted_bit_size, 0) def test_complains_about_invalid_cmd_tlm(self): - with self.assertRaisesRegex(AttributeError, f"Unknown type:OTHER"): + with self.assertRaisesRegex(AttributeError, "Unknown type:OTHER"): ObjectWriteConversion("OTHER", "TGT", "PKT") def test_writes_the_cmd_packet_and_returns_a_raw_block(self): diff --git a/openc3/python/test/models/test_metric_model.py b/openc3/python/test/models/test_metric_model.py index 5532fe2a69..5d453281cf 100644 --- a/openc3/python/test/models/test_metric_model.py +++ b/openc3/python/test/models/test_metric_model.py @@ -1,4 +1,3 @@ - # Copyright 2024 OpenC3, Inc. # All Rights Reserved. # @@ -15,71 +14,68 @@ # This file may also be used under the terms of a commercial license # if purchased from OpenC3, Inc. -import time import unittest -import fakeredis import unittest.mock from test.test_helper import * from openc3.models.metric_model import MetricModel -from openc3.conversions.generic_conversion import GenericConversion class TestMetricModel(unittest.TestCase): def setUp(self): - mock_redis(self) + mock_redis(self) def test_returns_all_the_metrics(self): - model = MetricModel(name= "foo", scope= "scope", values= {"test" : {"value" : 5}}) - model.create(force= True) - all_metrics = MetricModel.all(scope= "scope") + model = MetricModel(name="foo", scope="scope", values={"test": {"value": 5}}) + model.create(force=True) + all_metrics = MetricModel.all(scope="scope") self.assertEqual(all_metrics["foo"]["values"]["test"]["value"], (5)) def test_encodes_all_the_input_parameters(self): - model = MetricModel(name= "foo", scope= "scope", values= {"test" : {"value" : 5}}) + model = MetricModel(name="foo", scope="scope", values={"test": {"value": 5}}) json = model.as_json() self.assertEqual(json["name"], ("foo")) def test_gets_by_name_in_scope(self): - MetricModel(name= "baz", scope= "scope", values= {"test ": {"value" :6}}) - result = MetricModel.get(name= "baz", scope= "scope") - self.assertIsNone(result) #self.assertEqual(result['name'], ('baz')) + MetricModel(name="baz", scope="scope", values={"test ": {"value": 6}}) + result = MetricModel.get(name="baz", scope="scope") + self.assertIsNone(result) # self.assertEqual(result['name'], ('baz')) def test_destroys_by_name_in_scope(self): - MetricModel(name= "baz", scope= "scope", values= {"test ": {"value" :6}}) - MetricModel(name= "bOz", scope= "scope", values= {"test ": {"value" :6}}) - MetricModel.destroy(scope= 'scope', name= 'baz') - result = MetricModel.get(name= "baz", scope= "scope") + MetricModel(name="baz", scope="scope", values={"test ": {"value": 6}}) + MetricModel(name="bOz", scope="scope", values={"test ": {"value": 6}}) + MetricModel.destroy(scope="scope", name="baz") + result = MetricModel.get(name="baz", scope="scope") self.assertIsNone(result) def test_returns_all_names(self): - MetricModel(name= 'baz', scope= "scope", values= {"test ": {"value" :6}}) - result = MetricModel.names(scope= "scope") - self.assertListEqual(result, []) #self.assertEqual(result[0], ('baz')) + MetricModel(name="baz", scope="scope", values={"test ": {"value": 6}}) + result = MetricModel.names(scope="scope") + self.assertListEqual(result, []) # self.assertEqual(result[0], ('baz')) def test_returns_redis_metrics_from_store_and_ephemeral_store(self): values = { - 'connected_clients' : {'value' : 37}, - 'used_memory_rss' : {'value' : 0}, - 'total_commands_processed' : {'value' : 0}, - 'instantaneous_ops_per_sec' : {'value' : 0}, - 'instantaneous_input_kbps' : {'value' : 0}, - 'instantaneous_output_kbps' : {'value' : 0}, - 'latency_percentiles_usec_hget': {'value' : '1,2'} + "connected_clients": {"value": 37}, + "used_memory_rss": {"value": 0}, + "total_commands_processed": {"value": 0}, + "instantaneous_ops_per_sec": {"value": 0}, + "instantaneous_input_kbps": {"value": 0}, + "instantaneous_output_kbps": {"value": 0}, + "latency_percentiles_usec_hget": {"value": "1,2"}, } - model = MetricModel(name= "all", scope= "scope", values= {"test" : {"value" : 7}}) - model.create(force= True) + model = MetricModel(name="all", scope="scope", values={"test": {"value": 7}}) + model.create(force=True) json = {} - json['name'] = 'all' - json['values'] = values - MetricModel.set(json, scope= 'scope') + json["name"] = "all" + json["values"] = values + MetricModel.set(json, scope="scope") # awaiting FakeRedis support for the server INFO command # allow(openc3.Store.instance).to receive(:info) do - # values + # values # allow(openc3.EphemeralStore.instance).to receive(:info) do - # values + # values self.assertRaises(Exception, MetricModel.redis_metrics) - #self.assertEqual(result.empty?, (False)) - #self.assertEqual(result['redis_connected_clients_total']['value'], (37)) + # self.assertEqual(result.empty?, (False)) + # self.assertEqual(result['redis_connected_clients_total']['value'], (37)) diff --git a/openc3/python/test/models/test_secret_model.py b/openc3/python/test/models/test_secret_model.py index 5edb4b6242..4cbac3c3b1 100644 --- a/openc3/python/test/models/test_secret_model.py +++ b/openc3/python/test/models/test_secret_model.py @@ -1,4 +1,3 @@ - # Copyright 2024 OpenC3, Inc. # All Rights Reserved. # @@ -16,34 +15,32 @@ # This file may also be used under the terms of a commercial license # if purchased from OpenC3, Inc. -import time import unittest import unittest.mock from test.test_helper import * from openc3.models.secret_model import SecretModel -from openc3.conversions.generic_conversion import GenericConversion class TestSecretModel(unittest.TestCase): def setUp(self): - mock_redis(self) + mock_redis(self) def test_creates_new(self): - model = SecretModel(name= 'secret', value= 'tacit', scope= 'DEFAULT') + model = SecretModel(name="secret", value="tacit", scope="DEFAULT") self.assertIsInstance(model, SecretModel) def test_self_get(self): - name = SecretModel.get(name= 'secret', scope= 'DEFAULT') - self.assertIsNone(name) # eq('secret') + name = SecretModel.get(name="secret", scope="DEFAULT") + self.assertIsNone(name) # eq('secret') def test_self_all(self): - all_secrets = SecretModel.all(scope= 'DEFAULT') - self.assertEqual(all_secrets, {}) # eq('secret') + all_secrets = SecretModel.all(scope="DEFAULT") + self.assertEqual(all_secrets, {}) # eq('secret') def test_self_names(self): - names = SecretModel.names(scope= 'DEFAULT') - self.assertEqual(names, []) # eq('secret') + names = SecretModel.names(scope="DEFAULT") + self.assertEqual(names, []) # eq('secret') def test_as_json(self): - model = SecretModel(name= 'secreter', value= 'silent', scope= 'DEFAULT') - self.assertEqual(model.as_json()['name'], ('secreter')) + model = SecretModel(name="secreter", value="silent", scope="DEFAULT") + self.assertEqual(model.as_json()["name"], ("secreter")) diff --git a/openc3/python/test/models/test_stash_model.py b/openc3/python/test/models/test_stash_model.py index 95ba462fd8..fc9d9f9930 100644 --- a/openc3/python/test/models/test_stash_model.py +++ b/openc3/python/test/models/test_stash_model.py @@ -1,4 +1,3 @@ - # Copyright 2024 OpenC3, Inc. # All Rights Reserved. # @@ -16,35 +15,33 @@ # This file may also be used under the terms of a commercial license # if purchased from OpenC3, Inc. -import time import unittest import unittest.mock from test.test_helper import * from openc3.models.stash_model import StashModel -from openc3.conversions.generic_conversion import GenericConversion class TestStashModel(unittest.TestCase): def setUp(self): - mock_redis(self) - setup_system() + mock_redis(self) + setup_system() def test_creates_new(self): - model = StashModel(name= 'sm', value= 'stash', scope= 'DEFAULT') + model = StashModel(name="sm", value="stash", scope="DEFAULT") self.assertIsInstance(model, StashModel) def test_self_get(self): - name = StashModel.get(name= 'sm', scope= 'DEFAULT') - self.assertIsNone(name) # eq('sm') + name = StashModel.get(name="sm", scope="DEFAULT") + self.assertIsNone(name) # eq('sm') def test_self_all(self): - all_stash = StashModel.all(scope= 'DEFAULT') - self.assertEqual(all_stash, {}) # eq('sm') + all_stash = StashModel.all(scope="DEFAULT") + self.assertEqual(all_stash, {}) # eq('sm') def test_self_names(self): - names = StashModel.names(scope= 'DEFAULT') - self.assertEqual(names, []) # eq('sm') + names = StashModel.names(scope="DEFAULT") + self.assertEqual(names, []) # eq('sm') def test_as_json(self): - model = StashModel(name= 'sm', value= 'stashef', scope= 'DEFAULT') - self.assertEqual(model.as_json()['name'], ('sm')) + model = StashModel(name="sm", value="stashef", scope="DEFAULT") + self.assertEqual(model.as_json()["name"], ("sm"))