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.
		
		
		
		
		
			
		
			
				
	
	
		
			66 lines
		
	
	
		
			1.7 KiB
		
	
	
	
		
			Python
		
	
			
		
		
	
	
			66 lines
		
	
	
		
			1.7 KiB
		
	
	
	
		
			Python
		
	
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)
 | 
						|
		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']
 | 
						|
		print features
 | 
						|
		print app
 | 
						|
		label	= CONFIG['learner']['anomalies']['label']
 | 
						|
		x = lhandler.learn(data,'label',app,features,label)
 | 
						|
		print x
 | 
						|
		
 | 
						|
 | 
						|
if __name__ == '__main__' :
 | 
						|
	unittest.main() |