Skip to content
Browse files

Add a simple test that just copies files over

Tests are failing on jenkins although they pass locally because the seed model
is not found. But I'm copying the seed model!

Let's add a very simple test that basically just copies over the seed model so
that we can see the failure (if any) in the console.

Also switch the default test run over to this simple test
  • Loading branch information...
shankari committed Feb 24, 2018
1 parent 942d0c4 commit b4b578e9efe8259641963fe15c51ac357a7afad9
@@ -0,0 +1,105 @@
from __future__ import print_function
from __future__ import unicode_literals
from __future__ import division
from __future__ import absolute_import
#Standard imports
from future import standard_library
from builtins import str
from builtins import *
import unittest
import json
import arrow
import logging
import numpy as np
import os
from datetime import datetime, timedelta

# Our imports
import emission.core.get_database as edb
from emission.core.wrapper.user import User
from emission.core.wrapper.client import Client
import emission.core.wrapper.localdate as ecwl
import emission.core.wrapper.location as ecwlo
import emission.core.wrapper.section as ecws
import emission.core.wrapper.entry as ecwe
import emission.core.wrapper.modeprediction as ecwm
import emission.core.wrapper.motionactivity as ecwma

import emission.tests.common as etc
import as enufc

import emission.analysis.classification.inference.mode.pipeline as pipeline
import as esta
import as esda
import as esds

Create some outliers and make sure they're stripped out
class TestPipeline(unittest.TestCase):
def setUp(self):
# Thanks to M&J for the number!
self.copied_model_path = etc.copy_dummy_seed_for_inference()
dataFile = "emission/tests/data/real_examples/shankari_2016-08-10"
start_ld = ecwl.LocalDate({'year': 2016, 'month': 8, 'day': 9})
end_ld = ecwl.LocalDate({'year': 2016, 'month': 8, 'day': 10})
cacheKey = "diary/trips-2016-08-10"
etc.setupRealExample(self, dataFile)
# Default intake pipeline now includes mode inference
# this is correct in general, but causes errors while testing the mode inference
# because then that step is effectively run twice. This code
# rolls back the results of running the mode inference as part of the
# pipeline and allows us to correctly test the mode inference pipeline again.
pipeline.del_objects_after(self.testUUID, 0, is_dry_run=False)
self.pipeline = pipeline.ModeInferencePipeline()

def tearDown(self):
logging.debug("Clearing related databases")

def clearRelatedDb(self):
edb.get_timeseries_db().delete_many({"user_id": self.testUUID})
edb.get_analysis_timeseries_db().delete_many({"user_id": self.testUUID})
edb.get_usercache_db().delete_many({"user_id": self.testUUID})

def testFeatureGenWithOnePoint(self):
# ensure that the start and end datetimes are the same, since the average calculation uses
# the total distance and the total duration
ts = esta.TimeSeries.get_time_series(self.testUUID)
trackpoint1 = ecwlo.Location({u'coordinates': [0,0], 'type': 'Point'})
ts.insert_data(self.testUUID, "analysis/recreated_location", trackpoint1)
testSeg = ecws.Section({"start_loc": trackpoint1,
"end_loc": trackpoint1,
"distance": 500,
"sensed_mode": 1,
"duration": 150,
"_id": 2,
testSegEntry = ecwe.Entry.create_entry(self.testUUID, "analysis/cleaned_section", testSeg)
d =
m = testSegEntry.metadata
enufc.expand_start_end_data_times(d, m)
testSegEntry["data"] = d
testSegEntry["metadata"] = m
inserted_id = ts.insert(testSegEntry)
featureMatrix = np.zeros([1, len(self.pipeline.featureLabels)])
resultVector = np.zeros(1)
self.pipeline.updateFeatureMatrixRowWithSection(featureMatrix, 0, testSegEntry)
logging.debug("featureMatrix = %s" % featureMatrix)
self.assertEqual(np.count_nonzero(featureMatrix[0][5:16]), 0)
self.assertEqual(np.count_nonzero(featureMatrix[0][19:21]), 0)

if __name__ == '__main__':
@@ -214,8 +214,18 @@ def set_analysis_config(key, value):

def copy_dummy_seed_for_inference():
import shutil
import os

seed_json_source = "emission/tests/data/seed_model_from_test_data.json"
seed_json_dest = "seed_model.json"
shutil.copyfile(seed_json_source, seed_json_dest)
result = shutil.copyfile(seed_json_source, seed_json_dest)
logging.debug("Copied file %s -> %s with result %s" % (seed_json_source, seed_json_dest, result))

assert os.path.exists(seed_json_dest), "File %s not found after copy" % seed_json_dest

# Validation
import emission.analysis.classification.inference.mode.pipeline as eacimp
mip = eacimp.ModeInferencePipeline()

return seed_json_dest
@@ -6,4 +6,4 @@ set -e
# + ./
# rm: missing operand
# Try `rm --help' for more information.
PYTHONPATH=. python -m unittest discover -s emission/tests -p Test*;
PYTHONPATH=. python -m unittest discover -s emission/individual_tests -p Test*;

0 comments on commit b4b578e

Please sign in to comment.
You can’t perform that action at this time.