smart-top/test/TestML.py

from utils import transport
from utils.ml import ML, AnomalyDetection
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)
		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)
		x_ = ML.Extract(['cpu_usage','memory_usage'], 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'][1]
		lhandler = AnomalyDetection()
		features = CONFIG['learner']['anomalies']['features']
		label	= CONFIG['learner']['anomalies']['label']
		x = lhandler.learn(data,'label',app,features,label)
		print x
		

if __name__ == '__main__' :
	unittest.main()