Refactored dotmap.get_all_features() to take filenames

openaddr/dotmap.py

@@ -21,17 +21,18 @@ def main():
           '-X', '-n', 'OpenAddresses YYYY-MM-DD', '-f', '-o', '/tmp/openaddresses.mbtiles'
 
     tippecanoe = Popen(cmd, stdin=PIPE, bufsize=1)
+    zip_filenames = (fn for (_, fn, _) in iterate_local_processed_files(runs))
 
-    for feature in get_all_features(runs):
+    for feature in get_all_features(zip_filenames):
         print(json.dumps(feature), file=tippecanoe.stdin)
 
     tippecanoe.stdin.close()
     tippecanoe.wait()
 
-def get_all_features(runs):
+def get_all_features(zip_filenames):
     ''' Generate a stream of all locations as GeoJSON features.
     '''
-    for (_, fn, _) in iterate_local_processed_files(runs):
+    for fn in zip_filenames:
         print(fn, file=stderr)
         zipfile = ZipFile(fn, mode='r')
         for filename in zipfile.namelist():

openaddr/tests/dotmap.py

@@ -0,0 +1,56 @@
+from shutil import rmtree
+from os.path import join
+from tempfile import mkdtemp
+from zipfile import ZipFile
+import unittest
+
+import mock
+
+from ..dotmap import get_all_features
+
+class TestDotmap (unittest.TestCase):
+
+    def setUp(self):
+        self.test_dir = mkdtemp()
+        self.zipfiles = list()
+
+        self.zipfiles.append(join(self.test_dir, 'file1.zip'))
+        zf = ZipFile(self.zipfiles[-1], 'w')
+        zf.writestr('README.txt', 'Good times')
+        zf.writestr('stuff.csv', 'LAT,LON\n0,0\n37.804319,-122.271210\n')
+        zf.close()
+
+        self.zipfiles.append(join(self.test_dir, 'file2.zip'))
+        zf = ZipFile(self.zipfiles[-1], 'w')
+        zf.writestr('README.txt', 'Good times')
+        zf.writestr('stuff.csv', 'LON,LAT\n0,0\n-122.413729,37.775641\n')
+        zf.close()
+
+    def tearDown(self):
+        rmtree(self.test_dir)
+
+    def test_get_all_features_no_runs(self):
+        features = list(get_all_features(self.zipfiles[:0]))
+        self.assertEqual(len(features), 0)
+
+    def test_get_all_features_one_run(self):
+        features = list(get_all_features(self.zipfiles[:1]))
+
+        p1, p2 = [f['geometry']['coordinates'] for f in features]
+        self.assertAlmostEqual(p1[0],    0.0)
+        self.assertAlmostEqual(p1[1],    0.0)
+        self.assertAlmostEqual(p2[0], -122.271210)
+        self.assertAlmostEqual(p2[1],   37.804319)
+
+    def test_get_all_features_two_runs(self):
+        features = list(get_all_features(self.zipfiles[:2]))
+
+        p1, p2, p3, p4 = [f['geometry']['coordinates'] for f in features]
+        self.assertAlmostEqual(p1[0],    0.0)
+        self.assertAlmostEqual(p1[1],    0.0)
+        self.assertAlmostEqual(p2[0], -122.271210)
+        self.assertAlmostEqual(p2[1],   37.804319)
+        self.assertAlmostEqual(p3[0],    0.0)
+        self.assertAlmostEqual(p3[1],    0.0)
+        self.assertAlmostEqual(p4[0], -122.413729)
+        self.assertAlmostEqual(p4[1],   37.775641)

test.py

@@ -21,6 +21,7 @@
 from openaddr.tests.conform import TestConformCli, TestConformTransforms, TestConformMisc, TestConformCsv
 from openaddr.tests.expand import TestExpand
 from openaddr.tests.render import TestRender
+from openaddr.tests.dotmap import TestDotmap
 from openaddr.tests.util import TestEsri2GeoJSON
 from openaddr.tests.summarize import TestSummarizeFunctions
 from openaddr.tests.ci import TestHook, TestRuns, TestWorker, TestBatch, TestObjects, TestCollect