added unittest for scrapy shell command and fixed flake8 error

This commit is contained in:
Alex 2022-12-13 21:37:17 -08:00
parent 8b09b0e0d7
commit c2de9372a2
2 changed files with 12 additions and 0 deletions

View File

@ -80,6 +80,7 @@ def install_reactor(reactor_path, event_loop_path=None):
with suppress(error.ReactorAlreadyInstalledError):
installer()
def set_asyncio_event_loop(event_loop_path):
"""Sets and returns the event loop with specified import path."""
policy = get_asyncio_event_loop_policy()

View File

@ -115,3 +115,14 @@ class ShellTest(ProcessTest, SiteTest, unittest.TestCase):
errcode, out, err = yield self.execute([url, '-c', 'item'], check_code=False)
self.assertEqual(errcode, 1, out or err)
self.assertIn(b'DNS lookup failed', err)
@defer.inlineCallbacks
def test_shell_fetch_async(self):
reactor_path = "twisted.internet.asyncioreactor.AsyncioSelectorReactor"
url = self.url('/html')
code = f"fetch('{url}')"
args = ["-c", code, "--set", f"TWISTED_REACTOR={reactor_path}"]
_, _, err = yield self.execute(args, check_code=True)
self.assertNotIn(
b"RuntimeError: There is no current event loop in thread", err
)