You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
84 lines
2.2 KiB
Python
84 lines
2.2 KiB
Python
from utils import transport
|
|
from utils.ml import ML, AnomalyDetection, AnalyzeAnomaly
|
|
import unittest
|
|
import json
|
|
import os
|
|
|
|
|
|
path = os.environ['MONITOR_CONFIG_PATH']
|
|
f = open(path)
|
|
CONFIG = json.loads( f.read())
|
|
f.close()
|
|
factory = transport.DataSourceFactory()
|
|
|
|
class TestML(unittest.TestCase):
|
|
def setUp(self):
|
|
|
|
ref = CONFIG['store']['class']['read']
|
|
p = CONFIG['store']['args']
|
|
p['qid'] = ['apps']
|
|
self.greader = factory.instance(type=ref,args=p)
|
|
def test_has_date(self):
|
|
r = self.greader.read()
|
|
|
|
self.assertTrue(r)
|
|
def test_Filter(self):
|
|
r = self.greader.read()
|
|
r = r['apps']
|
|
#
|
|
# To make this test case extensible we need to pull apps from the configuration
|
|
#
|
|
app = CONFIG['monitor']['processes']['config']['apps'][0]
|
|
x = ML.Filter('label',app,r)
|
|
app = ML.CleanupName(app)
|
|
for row in x:
|
|
self.assertTrue(row['label'] == app)
|
|
def test_Extract(self):
|
|
r = self.greader.read()
|
|
r = r['apps']
|
|
app = CONFIG['monitor']['processes']['config']['apps'][0]
|
|
x = ML.Filter('label',app,r)
|
|
features = CONFIG['learner']['anomalies']['features']
|
|
self.assertTrue(features)
|
|
x_ = ML.Extract(features, x)
|
|
|
|
self.assertTrue (len (x) == len(x_))
|
|
pass
|
|
def test_Learn(self):
|
|
ref = CONFIG['store']['class']['read']
|
|
p = CONFIG['store']['args']
|
|
greader = factory.instance(type=ref,args=p)
|
|
|
|
data = greader.read()
|
|
|
|
data = data['apps']
|
|
app = CONFIG['monitor']['processes']['config']['apps'][0]
|
|
lhandler = AnomalyDetection()
|
|
features = CONFIG['learner']['anomalies']['features']
|
|
label = CONFIG['learner']['anomalies']['label']
|
|
x = lhandler.learn(data,'label',app,features,label)
|
|
|
|
|
|
def test_Predict(self):
|
|
ref = CONFIG['store']['class']['read']
|
|
p = CONFIG['store']['args']
|
|
greader = factory.instance(type=ref,args=p)
|
|
data = greader.read()
|
|
if 'learn' in data :
|
|
info = data['learn']
|
|
|
|
app = CONFIG['monitor']['processes']['config']['apps'][0]
|
|
|
|
lhandler = AnalyzeAnomaly()
|
|
features = CONFIG['learner']['anomalies']['features']
|
|
label = CONFIG['learner']['anomalies']['label']
|
|
#x = lhandler.learn(data,'label',app,features,label)
|
|
data = data['apps']
|
|
xo = ML.Filter('label',app,data)
|
|
info = ML.Filter('label',app,info)
|
|
|
|
lhandler.predict(xo,info[0])
|
|
|
|
if __name__ == '__main__' :
|
|
unittest.main()
|