smart-top/test/TestML.py

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()