add unittests

.github/workflows/unit_tests.yml

@@ -4,7 +4,7 @@ on:
     branches:
       - dev
     paths-ignore:
-      - 'ovos_bus_client/version.py'
+      - 'ocp_pipeline/version.py'
       - 'examples/**'
       - '.github/**'
       - '.gitignore'
@@ -17,7 +17,7 @@ on:
     branches:
       - master
     paths-ignore:
-      - 'ovos_bus_client/version.py'
+      - 'ocp_pipeline/version.py'
       - 'requirements/**'
       - 'examples/**'
       - '.github/**'
@@ -33,7 +33,7 @@ jobs:
   unit_tests:
     strategy:
       matrix:
-        python-version: [ 3.7, 3.8, 3.9, '3.10']
+        python-version: [3.9]
     runs-on: ubuntu-latest
     timeout-minutes: 15
     steps:
@@ -52,13 +52,10 @@ jobs:
           pip install -e .
       - name: Install test dependencies
         run: |
-          pip install -r test/requirements.txt
+          pip install -r tests/requirements.txt
       - name: Run unittests
         run: |
-          pytest --cov=ovos_adapt --cov-report=xml test/unittests
-          # NOTE: additional pytest invocations should also add the --cov-append flag
-          #       or they will overwrite previous invocations' coverage reports
-          #       (for an example, see OVOS Skill Manager's workflow)
+          pytest --cov=ocp_pipeline --cov-report=xml ./tests
       - name: Upload coverage
         if: "${{ matrix.python-version == '3.9' }}"
         uses: codecov/codecov-action@v3

requirements.txt

@@ -1 +1,2 @@
-ovos-workshop
+ovos-workshop
+ovos-classifiers

tests/requirements.txt

@@ -0,0 +1,5 @@
+coveralls>=1.8.2
+flake8>=3.7.9
+pytest>=5.2.4
+pytest-cov>=2.8.1
+cov-core>=1.15.0

tests/test_ocp.py

@@ -0,0 +1,190 @@
+import os.path
+import unittest
+from unittest.mock import patch, Mock
+
+from ovos_classifiers.skovos.features import ClassifierProbaVectorizer
+from sklearn.pipeline import FeatureUnion
+import ocp_pipeline.opm
+from ovos_bus_client.message import Message
+from ovos_utils.ocp import MediaType
+from ocp_pipeline.opm import OCPFeaturizer, OCPPipelineMatcher
+from ovos_utils.log import LOG
+
+
+class TestOCPFeaturizer(unittest.TestCase):
+
+    def setUp(self):
+        self.featurizer = OCPFeaturizer()
+
+    @patch('os.path.isfile', return_value=True)
+    @patch('ovos_classifiers.skovos.features.KeywordFeaturesVectorizer.load_entities')
+    @patch.object(LOG, 'info')
+    def test_load_csv_with_existing_file(self, mock_log_info, mock_load_entities, mock_isfile):
+        csv_path = "existing_file.csv"
+        self.featurizer.load_csv([csv_path])
+        mock_isfile.assert_called_with(csv_path)
+        mock_load_entities.assert_called_with(csv_path)
+        mock_log_info.assert_called_with(f"Loaded OCP keywords: {csv_path}")
+
+    @patch.object(LOG, 'error')
+    def test_load_csv_with_nonexistent_file(self, mock_log_error):
+        csv_path = "nonexistent_file.csv"
+        self.featurizer.load_csv([csv_path])
+        mock_log_error.assert_called_with(f"Requested OCP entities file does not exist? {csv_path}")
+
+    @patch.object(FeatureUnion, 'transform', return_value='mock_transform_result')
+    def test_transform(self, mock_transform):
+        self.featurizer.clf_feats = Mock(spec=ClassifierProbaVectorizer)
+        result = self.featurizer.transform(["example_text"])
+        mock_transform.assert_called_with(["example_text"])
+        self.assertEqual(result, 'mock_transform_result')
+
+
+class TestOCPPipelineNoClassifierMatcher(unittest.TestCase):
+
+    def setUp(self):
+        config = {
+            "experimental_media_classifier": False,
+            "experimental_binary_classifier": False,
+            "entity_csvs": [
+                os.path.dirname(ocp_pipeline.opm.__file__) + "/models/ocp_entities_v0.csv"
+            ]}
+        self.ocp = OCPPipelineMatcher(config=config)
+
+    def test_match_high(self):
+        result = self.ocp.match_high(["play metallica"], "en-us")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_intents')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_high_with_invalid_input(self):
+        result = self.ocp.match_high(["put on some music"], "en-us")
+        self.assertIsNone(result)
+
+    def test_match_medium(self):
+        result = self.ocp.match_medium(["put on some movie"], "en-us")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_media')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_medium_with_invalid_input(self):
+        result = self.ocp.match_medium(["i wanna hear metallica"], "en-us")
+        self.assertIsNone(result)
+
+    def test_match_fallback(self):
+        result = self.ocp.match_fallback(["i want music"], "en-us")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_fallback')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_fallback_with_invalid_input(self):
+        result = self.ocp.match_fallback(["do the thing"], "en-us")
+        self.assertIsNone(result)
+
+    def test_predict(self):
+        self.assertTrue(self.ocp.is_ocp_query("play a song", "en-us")[0])
+        self.assertTrue(self.ocp.is_ocp_query("play a movie", "en-us")[0])
+        self.assertTrue(self.ocp.is_ocp_query("play a podcast", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("tell me a joke", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("who are you", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("you suck", "en-us")[0])
+
+    def test_predict_prob(self):
+        noise = "hglisjerhksrtjhdgsf"
+        self.assertEqual(self.ocp.classify_media(f"play {noise} music", "en-us")[0], MediaType.MUSIC)
+        self.assertIsInstance(self.ocp.classify_media(f"play music {noise}", "en-us")[1], float)
+        self.assertEqual(self.ocp.classify_media(f"play {noise} movie soundtrack", "en-us")[0], MediaType.MUSIC)
+        self.assertEqual(self.ocp.classify_media(f"play movie {noise}", "en-us")[0], MediaType.MOVIE)
+        self.assertEqual(self.ocp.classify_media(f"play silent {noise} movie", "en-us")[0], MediaType.SILENT_MOVIE)
+        self.assertEqual(self.ocp.classify_media(f"play {noise} black and white movie", "en-us")[0],
+                         MediaType.BLACK_WHITE_MOVIE)
+        self.assertEqual(self.ocp.classify_media(f"play short {noise} film", "en-us")[0], MediaType.SHORT_FILM)
+        self.assertEqual(self.ocp.classify_media(f"play cartoons {noise}", "en-us")[0], MediaType.CARTOON)
+        self.assertEqual(self.ocp.classify_media(f"play {noise} episode", "en-us")[0], MediaType.VIDEO_EPISODES)
+        self.assertEqual(self.ocp.classify_media(f"play {noise} podcast", "en-us")[0], MediaType.PODCAST)
+        self.assertEqual(self.ocp.classify_media(f"play {noise} book", "en-us")[0], MediaType.AUDIOBOOK)
+        self.assertEqual(self.ocp.classify_media(f"play radio {noise} FM", "en-us")[0], MediaType.RADIO)
+        self.assertEqual(self.ocp.classify_media(f"read {noise}", "en-us")[0], MediaType.AUDIOBOOK)
+
+
+class TestOCPPipelineMatcher(unittest.TestCase):
+
+    def setUp(self):
+        config = {
+            "experimental_media_classifier": True,
+            "experimental_binary_classifier": True,
+            "entity_csvs": [
+                os.path.dirname(ocp_pipeline.opm.__file__) + "/models/ocp_entities_v0.csv"
+            ]}
+        self.ocp = OCPPipelineMatcher(config=config)
+
+    def test_match_high(self):
+        result = self.ocp.match_high(["play metallica"], "en-us")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_intents')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_high_with_invalid_input(self):
+        result = self.ocp.match_high(["put on some metallica"], "en-us")
+        self.assertIsNone(result)
+
+    def test_match_medium(self):
+        result = self.ocp.match_medium(["put on some metallica"], "en-us")
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_media')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_medium_with_invalid_input(self):
+        result = self.ocp.match_medium(["i wanna hear metallica"], "en-us")
+        self.assertIsNone(result)
+
+    def test_match_fallback(self):
+        result = self.ocp.match_fallback(["i wanna hear metallica"], "en-us")
+        print(result)
+        self.assertIsNotNone(result)
+        self.assertEqual(result.intent_service, 'OCP_fallback')
+        self.assertEqual(result.intent_type, 'ocp:play')
+
+    def test_match_fallback_with_invalid_input(self):
+        result = self.ocp.match_fallback(["do the thing"], "en-us")
+        self.assertIsNone(result)
+
+    def test_predict(self):
+        self.assertTrue(self.ocp.is_ocp_query("play a song", "en-us")[0])
+        self.assertTrue(self.ocp.is_ocp_query("play my morning jams", "en-us")[0])
+        self.assertTrue(self.ocp.is_ocp_query("i want to watch the matrix", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("tell me a joke", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("who are you", "en-us")[0])
+        self.assertFalse(self.ocp.is_ocp_query("you suck", "en-us")[0])
+
+    def test_predict_prob(self):
+        # "metallica" in csv dataset
+        self.ocp.config["classifier_threshold"] = 0.2
+        self.assertEqual(self.ocp.classify_media("play metallica", "en-us")[0], MediaType.MUSIC)
+        self.assertIsInstance(self.ocp.classify_media("play metallica", "en-us")[1], float)
+        self.ocp.config["classifier_threshold"] = 0.5
+        self.assertEqual(self.ocp.classify_media("play metallica", "en-us")[0], MediaType.GENERIC)
+        self.assertIsInstance(self.ocp.classify_media("play metallica", "en-us")[1], float)
+
+    @unittest.skip("TODO - classifiers needs retraining")
+    def test_predict_prob_with_unknown_entity(self):
+        # "klownevilus" not in the csv dataset
+        self.ocp.config["classifier_threshold"] = 0.2
+        self.assertEqual(self.ocp.classify_media("play klownevilus", "en-us")[0], MediaType.MUSIC)
+        self.assertIsInstance(self.ocp.classify_media("play klownevilus", "en-us")[1], float)
+        self.ocp.config["classifier_threshold"] = 0.5
+        self.assertEqual(self.ocp.classify_media("play klownevilus", "en-us")[0], MediaType.GENERIC)
+
+        self.ocp.config["classifier_threshold"] = 0.1
+        self.ocp.handle_skill_keyword_register(Message("", {
+            "skill_id": "fake",
+            "label": "movie_name",
+            "media_type": MediaType.MOVIE,
+            "samples": ["klownevilus"]
+        }))
+        # should be MOVIE not MUSIC  TODO fix me
+        self.assertEqual(self.ocp.classify_media("play klownevilus", "en-us")[0], MediaType.MOVIE)
+
+
+if __name__ == '__main__':
+    unittest.main()