tests for new class

src/py/crankshaft/test/test_clustering_kmeans.py

@@ -7,9 +7,10 @@ import numpy as np
 #
 # import sys
 # sys.modules['plpy'] = plpy
-from helper import plpy, fixture_file
+from helper import plpy, fixture_file, MockDBResponse
 import crankshaft.clustering as cc
 import json
+from collections import OrderedDict
 
 
 class KMeansTest(unittest.TestCase):
@@ -38,3 +39,38 @@ class KMeansTest(unittest.TestCase):
         self.assertEqual(len(np.unique(labels)), 2)
         self.assertEqual(len(c1), 20)
         self.assertEqual(len(c2), 20)
+
+
+class KMeansNonspatialTest(unittest.TestCase):
+    """Testing class for k-means non-spatial"""
+
+    def setUp(self):
+        plpy._reset()
+
+        # self.cluster_data = json.loads(
+        #     open(fixture_file('kmeans-nonspatial.json')).read())
+
+        self.params = {"subquery": "SELECT * FROM TABLE",
+                       "n_clusters": 5}
+
+    def test_kmeans_nonspatial(self):
+        """
+            test for k-means non-spatial
+        """
+        data_raw = [OrderedDict([("col1", [1, 1, 1, 4, 4, 4]),
+                                 ("col2", [2, 4, 0, 2, 4, 0]),
+                                 ("rowids", [1, 2, 3, 4, 5, 6])])]
+
+        data_obj = MockDBResponse(data_raw, [k for k in data_raw[0]
+                                             if k != 'rowids'])
+        plpy._define_result('select', data_obj)
+        clusters = cc.kmeans_nonspatial('subquery', ['col1', 'col2'], 4)
+
+        cl1 = clusters[0][1]
+        cl2 = clusters[3][1]
+
+        for idx, val in enumerate(clusters):
+            if idx < 3:
+                self.assertEqual(val[1], cl1)
+            else:
+                self.assertEqual(val[1], cl2)

src/py/crankshaft/test/test_clustering_moran.py

@@ -7,13 +7,13 @@ import numpy as np
 #
 # import sys
 # sys.modules['plpy'] = plpy
-from helper import plpy, fixture_file
+from helper import plpy, fixture_file, MockDBResponse
 
 import crankshaft.clustering as cc
 import crankshaft.pysal_utils as pu
 from crankshaft import random_seeds
 import json
-
+from collections import OrderedDict
 
 class MoranTest(unittest.TestCase):
     """Testing class for Moran's I functions"""
@@ -58,11 +58,14 @@ class MoranTest(unittest.TestCase):
 
     def test_moran_local(self):
         """Test Moran's I local"""
-        data = [{'id': d['id'],
-                 'attr1': d['value'],
-                 'neighbors': d['neighbors']} for d in self.neighbors_data]
+        data = [OrderedDict([('id', d['id']),
+                             ('attr1', d['value']),
+                             ('neighbors', d['neighbors'])])
+                for d in self.neighbors_data]
 
-        plpy._define_result('select', data)
+        db_resp = MockDBResponse(data)
+
+        plpy._define_result('select', db_resp)
         random_seeds.set_random_seeds(1234)
         result = cc.moran_local('subquery', 'value',
                                 'knn', 5, 99, 'the_geom', 'cartodb_id')