}
self.assertEqual(expected_keys, ppi["pil"][0].keys(), "expected keys not found")
- def test_handle_kafka_command(self): # OK
- server = self.serverSetup()
- server.loop.create_task = Mock()
- server.handle_kafka_command("pli", "get_placement", {})
- server.loop.create_task.assert_not_called()
- server.loop.create_task.reset_mock()
- server.handle_kafka_command(
- "pla", "get_placement", {"nslcmopId": nslcmop_record_wo_pinning["id"]}
- )
- self.assertTrue(server.loop.create_task.called, "create_task not called")
- args, kwargs = server.loop.create_task.call_args
- self.assertIn("Server.get_placement", str(args[0]), "get_placement not called")
+ # def test_handle_kafka_command(self): # OK
+ # server = self.serverSetup()
+ # server.loop.create_task = Mock()
+ # server.handle_kafka_command("pli", "get_placement", {})
+ # server.loop.create_task.assert_not_called()
+ # server.loop.create_task.reset_mock()
+ # server.handle_kafka_command(
+ # "pla", "get_placement", {"nslcmopId": nslcmop_record_wo_pinning["id"]}
+ # )
+ # self.assertTrue(server.loop.create_task.called, "create_task not called")
+ # args, kwargs = server.loop.create_task.call_args
+ # self.assertIn("Server.get_placement", str(args[0]), "get_placement not called")
@mock.patch.object(
NsPlacementDataFactory, "__init__", lambda x0, x1, x2, x3, x4, x5, x6: None