Update tests for spotifyconnect

amplipi/streams/spotify_connect.py

@@ -2,13 +2,14 @@
 
 import io
 import os
+import re
 import sys
 import subprocess
 import time
 from typing import ClassVar, Optional
 import yaml
 from amplipi import models, utils
-from .base_streams import PersistentStream, logger
+from .base_streams import PersistentStream, InvalidStreamField, logger
 from .. import tasks
 
 # Our subprocesses run behind the scenes, is there a more standard way to do this?
@@ -112,7 +113,7 @@ def _deactivate(self):
         self.proc2.kill()
       self.proc.communicate()
       self.proc2.communicate()
-    if self._log_file:
+    if self.proc and self._log_file:  # prevent checking _log_file when it may not exist, thanks validation!
       self._log_file.close()
     if self.src:
       try:
@@ -185,6 +186,11 @@ def send_cmd(self, cmd: str) -> None:
     else:
       raise NotImplementedError(f'Spotify command not supported: {cmd}')
 
+  def validate_stream(self, **kwargs):
+    NAME = r"[a-zA-Z0-9][A-Za-z0-9\- ]*[a-zA-Z0-9]"
+    if 'name' in kwargs and not re.fullmatch(NAME, kwargs['name']):
+      raise InvalidStreamField("name", "Invalid stream name")
+
   def sync_volume(self, volume: float) -> None:
     """ Set the volume of amplipi to the Spotify Connect stream"""
     if volume != self.last_volume:

tests/test_rest.py

@@ -53,7 +53,7 @@
   {"id": AP_STREAM_ID, "name": "AmpliPi", "type": "shairport", "ap2": False},
   {"id": P_STREAM_ID, "name": "Radio Station, needs user/pass/station-id", "type": "pandora",
    "user": "[email protected]", "password": "CHANGEME", "station": "CHANGEME"},
-  {"id": 1002, "name": "AmpliPi", "type": "spotify"},
+  {"id": 1002, "name": "AmpliPi", "type": "spotifyconnect"},
   {"id": 1003, "name": "Groove Salad", "type": "internetradio",
    "url": "http://ice6.somafm.com/groovesalad-32-aac", "logo": "https://somafm.com/img3/groovesalad-400.jpg"},
   {"id": 1004, "name": "AmpliPi", "type": "dlna"},
@@ -725,7 +725,7 @@ def pandora_users():
 
 
 def spotify_stream_ids():
-  return [s['id'] for s in base_config()['streams'] if s['type'] == 'spotify']
+  return [s['id'] for s in base_config()['streams'] if s['type'] == 'spotifyconnect']
 
 
 def spotify_names():
@@ -840,7 +840,7 @@ def test_create_pandora(client, user, valid):
 @pytest.mark.parametrize('name, valid', spotify_names())
 def test_create_spotify(client, name, valid):
   """ Try creating a Spotify stream """
-  m_and_k = {'name': name, 'type': 'spotify'}
+  m_and_k = {'name': name, 'type': 'spotifyconnect'}
   rv = client.post('/api/stream', json=m_and_k)
   # check that the stream has an id added to it and that all of the fields are still there
   if valid: