■ ■ ■ ■ ■ ■
plugin_server/py/plugin_service_test.py
| skipped 14 lines |
15 | 15 | | |
16 | 16 | | import time |
17 | 17 | | |
| 18 | + | from absl.testing import absltest |
18 | 19 | | import grpc_testing |
19 | 20 | | import ipaddr |
20 | 21 | | |
21 | 22 | | from google.protobuf import timestamp_pb2 |
22 | | - | from net.proto2.contrib.pyutil import compare |
23 | | - | from testing.pybase import googletest |
24 | 23 | | from tsunami.plugin_server.py import plugin_service |
25 | 24 | | from tsunami.plugin_server.py import tsunami_plugin |
26 | 25 | | from tsunami.proto import detection_pb2 |
| skipped 16 lines |
43 | 42 | | MAX_WORKERS = 1 |
44 | 43 | | |
45 | 44 | | |
46 | | - | class PluginServiceTest(googletest.TestCase): |
| 45 | + | class PluginServiceTest(absltest.TestCase): |
47 | 46 | | |
48 | | - | @classmethod |
49 | | - | def setUpClass(cls): |
50 | | - | super().setUpClass() |
51 | | - | cls.test_plugin = FakeVulnDetector() |
52 | | - | cls._time = grpc_testing.strict_fake_time(time.time()) |
53 | | - | cls._server = grpc_testing.server_from_dictionary( |
| 47 | + | def setUp(self): |
| 48 | + | super().setUp() |
| 49 | + | self.test_plugin = FakeVulnDetector() |
| 50 | + | self._time = grpc_testing.strict_fake_time(time.time()) |
| 51 | + | self._server = grpc_testing.server_from_dictionary( |
54 | 52 | | { |
55 | 53 | | _ServiceDescriptor: |
56 | 54 | | plugin_service.PluginServiceServicer( |
57 | | - | py_plugins=[cls.test_plugin], max_workers=MAX_WORKERS), |
58 | | - | }, cls._time) |
| 55 | + | py_plugins=[self.test_plugin], max_workers=MAX_WORKERS), |
| 56 | + | }, self._time) |
59 | 57 | | |
60 | | - | cls._channel = grpc_testing.channel( |
61 | | - | plugin_service_pb2.DESCRIPTOR.services_by_name.values(), cls._time) |
| 58 | + | self._channel = grpc_testing.channel( |
| 59 | + | plugin_service_pb2.DESCRIPTOR.services_by_name.values(), self._time) |
62 | 60 | | |
63 | | - | @classmethod |
64 | | - | def tearDownClass(cls): |
65 | | - | cls._channel.close() |
66 | | - | super().tearDownClass() |
| 61 | + | def tearDown(self): |
| 62 | + | self._channel.close() |
| 63 | + | super().tearDown() |
67 | 64 | | |
68 | 65 | | def test_run_plugins_registered_returns_valid_response(self): |
69 | 66 | | plugin_to_test = FakeVulnDetector() |
| skipped 34 lines |
104 | 101 | | request = plugin_service.ListPluginsRequest() |
105 | 102 | | rpc = self._server.invoke_unary_unary(_ListPluginsMethod, (), request, None) |
106 | 103 | | response, _, _, _ = rpc.termination() |
107 | | - | compare.assertProto2Equal( |
108 | | - | self, |
| 104 | + | self.assertEqual( |
109 | 105 | | plugin_service.ListPluginsResponse( |
110 | 106 | | plugins=[self.test_plugin.GetPluginDefinition()]), response) |
111 | 107 | | |
| skipped 54 lines |
166 | 162 | | # TODO(b/239628051): Add a failed VulnDetector class to test failed cases. |
167 | 163 | | |
168 | 164 | | if __name__ == '__main__': |
169 | | - | googletest.main() |
| 165 | + | absltest.main() |
170 | 166 | | |