chore(refactor): Move get_crawl_args function to caller

scrapyd/launcher.py

@@ -8,7 +8,24 @@
 
 from scrapyd import __version__
 from scrapyd.interfaces import IEnvironment, IJobStorage, IPoller
-from scrapyd.utils import get_crawl_args, native_stringify_dict
+from scrapyd.utils import native_stringify_dict, to_native_str
+
+
+def get_crawl_args(message):
+    """Return the command-line arguments to use for the scrapy crawl process
+    that will be started for this message
+    """
+    msg = message.copy()
+    args = [to_native_str(msg["_spider"])]
+    del msg["_project"], msg["_spider"]
+    settings = msg.pop("settings", {})
+    for k, v in native_stringify_dict(msg, keys_only=False).items():
+        args += ["-a"]
+        args += [f"{k}={v}"]
+    for k, v in native_stringify_dict(settings, keys_only=False).items():
+        args += ["-s"]
+        args += [f"{k}={v}"]
+    return args
 
 
 class Launcher(Service):

scrapyd/utils.py

@@ -76,36 +76,19 @@ def native_stringify_dict(dct_or_tuples, encoding="utf-8", *, keys_only=True):
     """
     d = {}
     for k, v in dct_or_tuples.items():
-        key = _to_native_str(k, encoding)
+        key = to_native_str(k, encoding)
         if keys_only:
             value = v
         elif isinstance(v, dict):
             value = native_stringify_dict(v, encoding=encoding, keys_only=keys_only)
         elif isinstance(v, list):
-            value = [_to_native_str(e, encoding) for e in v]
+            value = [to_native_str(e, encoding) for e in v]
         else:
-            value = _to_native_str(v, encoding)
+            value = to_native_str(v, encoding)
         d[key] = value
     return d
 
 
-def get_crawl_args(message):
-    """Return the command-line arguments to use for the scrapy crawl process
-    that will be started for this message
-    """
-    msg = message.copy()
-    args = [_to_native_str(msg["_spider"])]
-    del msg["_project"], msg["_spider"]
-    settings = msg.pop("settings", {})
-    for k, v in native_stringify_dict(msg, keys_only=False).items():
-        args += ["-a"]
-        args += [f"{k}={v}"]
-    for k, v in native_stringify_dict(settings, keys_only=False).items():
-        args += ["-s"]
-        args += [f"{k}={v}"]
-    return args
-
-
 def get_spider_list(project, runner=None, pythonpath=None, version=None):
     """Return the spider list from the given project, using the given runner"""
 
@@ -150,7 +133,7 @@ def get_spider_list(project, runner=None, pythonpath=None, version=None):
     return spiders
 
 
-def _to_native_str(text, encoding="utf-8", errors="strict"):
+def to_native_str(text, encoding="utf-8", errors="strict"):
     if isinstance(text, str):
         return text
     return text.decode(encoding, errors)

tests/test_launcher.py

@@ -0,0 +1,21 @@
+from scrapyd.launcher import get_crawl_args
+
+
+def test_get_crawl_args():
+    msg = {"_project": "lolo", "_spider": "lala"}
+
+    assert get_crawl_args(msg) == ["lala"]
+
+    msg = {"_project": "lolo", "_spider": "lala", "arg1": "val1"}
+    cargs = get_crawl_args(msg)
+
+    assert cargs == ["lala", "-a", "arg1=val1"]
+    assert all(isinstance(x, str) for x in cargs), cargs
+
+
+def test_get_crawl_args_with_settings():
+    msg = {"_project": "lolo", "_spider": "lala", "arg1": "val1", "settings": {"ONE": "two"}}
+    cargs = get_crawl_args(msg)
+
+    assert cargs == ["lala", "-a", "arg1=val1", "-s", "ONE=two"]
+    assert all(isinstance(x, str) for x in cargs), cargs