Add test predictUserLoad with 2 cases

Author: QuangTung97

__pycache__/prediction.cpython-35.pyc

@@ -3,8 +3,8 @@
 import MySQLdb
 
 
-def changeAlgorithm(new_avg, prev_avg, difference):
-    return new_avg
+def changeAlgorithm(new_avg, prev_avg):
+    return (new_avg + prev_avg) / 2
 
 
 def predictUserLoad(user, jsample_repo, pred_repo):
@@ -17,17 +17,8 @@ def predictUserLoad(user, jsample_repo, pred_repo):
         # As previous system statistic
         pred = pred_repo.get(user.uid)
 
-        # We want to adjust the value of average CPU for the user
-        # in the prediction table by using an algorithm that can
-        # adjust the value based on the difference
-        difference = stat.avg_cpu - pred.avg_cpu
-        new_avg_cpu = changeAlgorithm(stat.avg_cpu, pred.avg_cpu,
-                                      difference)
-
-        # Similiar with RAM
-        difference = stat.avg_ram - pred.avg_ram
-        new_avg_ram = changeAlgorithm(stat.avg_ram, pred.avg_ram,
-                                      difference)
+        new_avg_cpu = changeAlgorithm(stat.avg_cpu, pred.avg_cpu)
+        new_avg_ram = changeAlgorithm(stat.avg_ram, pred.avg_ram)
 
         # We check if the user have heavier CPU jobs
         # running in the system

__pycache__/utils.cpython-35.pyc

@@ -0,0 +1,84 @@
+from prediction import predictUserLoad
+from utils import *
+import unittest
+from mock import MagicMock
+
+
+class TestPredictionMethod(unittest.TestCase):
+    def setUp(self):
+        # Mock jSampleRepository
+        self.jsample_repo = jSampleRepository(None)
+
+        self.stat = jSampleStatistic(
+            avg_cpu=10.0, max_cpu=17.2,
+            avg_ram=20.0, max_ram=25.4,
+            run_time="2017-4-8 11:25:00"
+        )
+        self.jsample_repo.statistic = MagicMock(
+            return_value=self.stat
+        )
+
+        # Mock PredictionRepository
+        self.pred_repo = PredictionRepository(None)
+
+        self.user = User(1000, "user1", "server1")
+
+    def test_predictUserLoad_UserExist(self):
+        self.pred_repo.exist = MagicMock(
+            return_value=True
+        )
+
+        newuser = User(1000, "user1", "server2")
+
+        newstat = jSampleStatistic(
+            avg_cpu=15.0, max_cpu=20.3,
+            avg_ram=25.333, max_ram=30.1,
+            run_time="2017-4-8 10:00:00"
+        )
+        pred = Prediction(newuser, newstat)
+
+        self.pred_repo.get = MagicMock(
+            return_value=pred
+        )
+        self.pred_repo.update = MagicMock()
+        self.jsample_repo.deleteUidsEarlierThan = MagicMock()
+
+        predictUserLoad(self.user, self.jsample_repo,
+                        self.pred_repo)
+
+        self.pred_repo.get.assert_called_once_with(1000)
+
+        self.assertEqual(pred.avg_cpu, 12.5)
+        self.assertEqual(pred.max_cpu, 20.3)
+        self.assertEqual(pred.avg_ram, 22.666)
+        self.assertEqual(pred.max_ram, 30.1)
+        self.assertEqual(pred.last_used_server, "server1")
+        self.assertEqual(pred.last_login, "2017-4-8 11:25:00")
+
+    def test_predictUserLoad_UserNotExist(self):
+        self.pred_repo.exist = MagicMock(
+            return_value=False
+        )
+        self.pred_repo.add = MagicMock()
+
+        predictUserLoad(self.user, self.jsample_repo,
+                        self.pred_repo)
+
+        self.pred_repo.add.assert_called_once()
+
+        # pred_repo.add call with right args
+        pred_called = self.pred_repo.add.call_args[0][0]
+        pred = Prediction(self.user, self.stat)
+        self.assertEqual(pred_called.uid, pred.uid)
+        self.assertEqual(pred_called.username, pred.username)
+        self.assertEqual(pred_called.last_used_server,
+                         pred.last_used_server)
+        self.assertEqual(pred_called.last_login, pred.last_login)
+        self.assertEqual(pred_called.avg_cpu, pred.avg_cpu)
+        self.assertEqual(pred_called.max_cpu, pred.max_cpu)
+        self.assertEqual(pred_called.avg_ram, pred.avg_ram)
+        self.assertEqual(pred_called.max_ram, pred.max_ram)
+
+
+if __name__ == '__main__':
+    unittest.main()

prediction.py

@@ -10,22 +10,22 @@ def BytesToMB(bytes):
 # Black box testing
 class TestPsutil(unittest.TestCase):
     def test_cpu_count(self):
-        assert(psutil.cpu_count() == 4)
+        self.assertEqual(psutil.cpu_count(), 4)
 
     def test_cpu_percent(self):
         percent = psutil.cpu_percent()
-        assert(percent >= 0 and percent <= 400)
+        self.assertTrue(percent >= 0 and percent <= 400)
 
     def test_virtual_memory(self):
         total, available, percent, used, free, active, inactive,            \
             buff, cache, shared = psutil.virtual_memory()
 
-        assert(percent >= 0 and percent <= 100)
+        self.assertTrue(percent >= 0 and percent <= 100)
         available = BytesToMB(available)
         cache = BytesToMB(cache)
-        assert(cache >= 100 and cache < 2048)
-        assert(available >= 128)
-        assert(available <= 4096)
+        self.assertTrue(cache >= 100 and cache < 2048)
+        self.assertTrue(available >= 128)
+        self.assertTrue(available <= 4096)
 
     def test_disk_io_counters(self):
         read_count, write_count, read_bytes, write_bytes,                   \
@@ -34,35 +34,35 @@ def test_disk_io_counters(self):
             = psutil.disk_io_counters()
         disk_in = BytesToMB(read_bytes)
         disk_out = BytesToMB(write_bytes)
-        assert(disk_in < 10 * 1024)
-        assert(disk_in > 128)
-        assert(disk_out < 10 * 1024)
-        assert(disk_out > 128)
+        self.assertTrue(disk_in < 10 * 1024)
+        self.assertTrue(disk_in > 128)
+        self.assertTrue(disk_out < 10 * 1024)
+        self.assertTrue(disk_out > 128)
 
     def test_process_iter_and_uids(self):
         count = 0
         for proc in psutil.process_iter():
             count += 1
             real, effective, saved = proc.uids()
-            assert(real >= 0 and real <= 0xffff)
-            assert(effective >= 0 and effective <= 0xffff)
-            assert(saved >= 0 and saved <= 0xffff)
-        assert(count > 0)
+            self.assertTrue(real >= 0 and real <= 0xffff)
+            self.assertTrue(effective >= 0 and effective <= 0xffff)
+            self.assertTrue(saved >= 0 and saved <= 0xffff)
+        self.assertTrue(count > 0)
 
     def test_psutil_process_functions(self):
         for proc in psutil.process_iter():
             real, effective, saved = proc.uids()
             if real < 1000:
                 continue
-            assert(proc.pid <= 0xffff)
-            assert(proc.ppid() >= 0 and proc.ppid() <= 0xffff)
-            assert(isinstance(proc.name(), str))
-            assert(isinstance(proc.username(), str))
+            self.assertTrue(proc.pid <= 0xffff)
+            self.assertTrue(proc.ppid() >= 0 and proc.ppid() <= 0xffff)
+            self.assertIsInstance(proc.name(), str)
+            self.assertIsInstance(proc.username(), str)
             timestamp = datetime.datetime                   \
                 .fromtimestamp(proc.create_time())          \
                 .strftime("%Y-%m-%d %H:%M:%S")
-            assert(isinstance(timestamp, str))
-            assert(isinstance(proc.cmdline(), list))
+            self.assertIsInstance(timestamp, str)
+            self.assertIsInstance(proc.cmdline(), list)
 
             rIO, wIO, rB, wB, read_chars, write_chars       \
                 = proc.io_counters()
@@ -76,12 +76,12 @@ def test_psutil_process_functions(self):
             rIO = BytesToMB(rIO)
             wIO = BytesToMB(wIO)
 
-            assert(rss <= 1024)
-            assert(vms <= 1024)
-            assert(cpu >= 0 and cpu <= 100)
-            assert(ram >= 0 and ram <= 100)
-            assert(rIO < 1024)
-            assert(wIO < 1024)
+            self.assertTrue(rss <= 1024)
+            self.assertTrue(vms <= 1024)
+            self.assertTrue(cpu >= 0 and cpu <= 100)
+            self.assertTrue(ram >= 0 and ram <= 100)
+            self.assertTrue(rIO < 1024)
+            self.assertTrue(wIO < 1024)
             break