RIC-642 related changes: REST subscription, rnib enhancements, symptomdata, rest...

[ric-plt/xapp-frame-py.git] / tests / test_symptomdata.py

diff --git a/tests/test_symptomdata.py b/tests/test_symptomdata.py
new file mode 100644 (file)
index 0000000..93a8fdb
--- /dev/null
+++ b/tests/test_symptomdata.py
@@ -0,0 +1,149 @@
+import os
+import requests
+from zipfile import ZipFile
+from requests import Response as response
+from ricxappframe.xapp_symptomdata import Symptomdata
+
+
+def new_loaddata(*args, **kwargs):
+    # Your custom testing override
+    return ""
+
+
+class MockResponse(object):
+    def __init__(self, reponse, jsonout):
+        self.status_code = response
+        self.url = 'http://lwsd.ricplt:8089/ric/v1/lwsd'
+        self.headers = {'Content-type': 'application/json'}
+        self.jsonout = jsonout
+
+    def json(self):
+        return self.jsonout
+
+    def raise_for_status(self):
+        return self.status_code
+
+
+class MockStat(object):
+    def __init__(self, st_ctime=None):
+        self.st_ctime = st_ctime
+
+
+class MockOs(object):
+    def __init__(self, walk=None, st_ctime=None):
+        self.walkresp = walk
+        self.st_ctime = st_ctime
+
+    def walk(self, path):
+        return self.walkresp
+
+    def stat(self, filename):
+        st = MockStat(self.st_ctime)
+        return st
+
+
+def test_symptomdata_subscribe(monkeypatch):
+    def mock_requests_post(uri, data, headers, proxies):
+        print("%s %s" % (uri, data))
+        return MockResponse(200, [{'service': 'xapp.service'}])
+
+    def mock_requests_get(uri, headers, proxies):
+        print("%s" % (uri))
+        return MockResponse(200, [])
+
+    # mock the http get and post
+    monkeypatch.setattr(requests, 'post', mock_requests_post)
+    monkeypatch.setattr(requests, 'get', mock_requests_get)
+
+    # this will return not found
+    s = Symptomdata("xapp", "xapp.ricxapp.service", "tmp", "http://lwsd.ricplt:8089/ric/v1/lwsd")
+    # stop timer loop
+    s.stop()
+    # make subscription
+    s.subscribe(None)
+    assert s.lwsdok is True
+
+
+def test_symptomdata_subscribe_exists(monkeypatch):
+    def mock_requests_get(uri, headers, proxies):
+        print("%s" % (uri))
+        return MockResponse(200, [{'service': 'xapp_other'}, {'service': 'xapp'}])
+
+    # mock the http get
+    monkeypatch.setattr(requests, 'get', mock_requests_get)
+
+    # this will return not found
+    s = Symptomdata("xapp", "xapp.ricxapp.service", "tmp", "http://lwsd.ricplt:8089/ric/v1/lwsd")
+    # stop timer loop
+    s.stop()
+    assert s.lwsdok is True
+
+
+def test_symptomdata_collect_time(monkeypatch):
+    myos = MockOs(walk=[('mydir', (), ('file1.csv', 'file2.csv', 'file3.txt', 'file.json'))], st_ctime=1647502471)
+
+    def mock_requests_get(uri, headers, proxies):
+        return MockResponse(200, [{'service': 'xapp_other'}, {'service': 'xapp'}])
+
+    def mock_os_walk(path):
+        return myos.walk(path)
+
+    def mock_os_stat(filename):
+        return myos.stat(filename)
+
+    def mock_zipfile_write(me, fromfile, tofile):
+        return
+
+    # mock the http get
+    monkeypatch.setattr(requests, 'get', mock_requests_get)
+    # mock the os walk
+    monkeypatch.setattr(os, 'walk', mock_os_walk)
+    # mock the os stat
+    monkeypatch.setattr(os, 'stat', mock_os_stat)
+
+    # mock the zipfile stat
+    monkeypatch.setattr(ZipFile, 'write', mock_zipfile_write)
+
+    # this will return not found
+    s = Symptomdata("xapp", "xapp.ricxapp.service", "tmp", "http://lwsd.ricplt:8089/ric/v1/lwsd")
+    # stop timer loop
+    s.stop()
+    assert s.lwsdok is True
+
+    zipfile = s.collect("zipfile.zip", (r'/tmp/csv/.*\.csv', r'/tmp/json/.*\.json'), 1647502470, 1647502570)
+    assert zipfile is not None
+
+
+def test_symptomdata_collect(monkeypatch):
+    myos = MockOs(walk=[('mydir', (), ('file1.csv', 'file2.csv', 'file3.txt', 'file.json'))], st_ctime=1647502471)
+
+    def mock_requests_get(uri, headers, proxies):
+        return MockResponse(200, [{'service': 'xapp_other'}, {'service': 'xapp'}])
+
+    def mock_os_walk(path):
+        return myos.walk(path)
+
+    def mock_os_stat(filename):
+        return myos.stat(filename)
+
+    def mock_zipfile_write(me, fromfile, tofile):
+        return
+
+    # mock the http get
+    monkeypatch.setattr(requests, 'get', mock_requests_get)
+    # mock the os walk
+    monkeypatch.setattr(os, 'walk', mock_os_walk)
+    # mock the os stat
+    monkeypatch.setattr(os, 'stat', mock_os_stat)
+
+    # mock the zipfile stat
+    monkeypatch.setattr(ZipFile, 'write', mock_zipfile_write)
+
+    # this will return not found
+    s = Symptomdata("xapp", "xapp.ricxapp.service", "tmp", "http://lwsd.ricplt:8089/ric/v1/lwsd")
+    # stop timer loop
+    s.stop()
+    assert s.lwsdok is True
+
+    zipfile = s.collect("zipfile.zip", ('/tmp/csv/.*.csv', '/tmp/json/.*.json'), 0, 0)
+    assert zipfile is not None