1 files changed, 89 insertions, 0 deletions
diff --git a/tests/mock_websocket.py b/tests/mock_websocket.py
new file mode 100644
index 0000000..ac64ed8
--- /dev/null
+++ b/tests/mock_websocket.py
@@ -0,0 +1,89 @@
+# ============LICENSE_START=======================================================
+# Copyright (c) 2018 AT&T Intellectual Property. All rights reserved.
+# ================================================================================
+# 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
+#
+#      http://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.
+# ============LICENSE_END=========================================================
+#
+# ECOMP is a trademark and service mark of AT&T Intellectual Property.
+"""mocking for the websocket - for testing of policy-engine"""
+
+import json
+import time
+
+from policyhandler.policy_consts import POLICY_NAME
+from policyhandler.policy_receiver import (LOADED_POLICIES, POLICY_VER,
+                                           REMOVED_POLICIES)
+
+from .mock_policy_engine import MockPolicyEngine
+from .mock_settings import Settings
+
+
+class MockWebSocket(object):
+    """Mock websocket"""
+    on_message = None
+
+    @staticmethod
+    def send_notification(updated_indexes):
+        """fake notification through the web-socket"""
+        if not MockWebSocket.on_message:
+            return
+        message = {
+            LOADED_POLICIES: [
+                {POLICY_NAME: "{0}.{1}.xml".format(
+                    MockPolicyEngine.get_policy_id(policy_index), policy_index + 1),
+                 POLICY_VER: str(policy_index + 1)}
+                for policy_index in updated_indexes or []
+            ],
+            REMOVED_POLICIES : []
+        }
+        message = json.dumps(message)
+        Settings.logger.info("send_notification: %s", message)
+        MockWebSocket.on_message(None, message)
+
+    @staticmethod
+    def enableTrace(yes_no):
+        """ignore"""
+        pass
+
+    class MockSocket(object):
+        """Mock websocket"""
+        def __init__(self):
+            self.connected = True
+
+    class WebSocketApp(object):
+        """Mocked WebSocketApp"""
+        def __init__(self, web_socket_url,
+                     on_open=None, on_message=None, on_close=None, on_error=None, on_pong=None):
+            self.web_socket_url = web_socket_url
+            self.on_open = on_open
+            self.on_message = MockWebSocket.on_message = on_message
+            self.on_close = on_close
+            self.on_error = on_error
+            self.on_pong = on_pong
+            self.sock = MockWebSocket.MockSocket()
+            Settings.logger.info("MockWebSocket for: %s", self.web_socket_url)
+
+        def run_forever(self, sslopt=None):
+            """forever in the loop"""
+            Settings.logger.info("MockWebSocket run_forever with sslopt=%s...",
+                                 json.dumps(sslopt))
+            counter = 0
+            while self.sock.connected:
+                counter += 1
+                Settings.logger.info("MockWebSocket sleep %s...", counter)
+                time.sleep(5)
+            Settings.logger.info("MockWebSocket exit %s", counter)
+
+        def close(self):
+            """close socket"""
+            self.sock.connected = False