Skip to content

Commit

Permalink
tests: Remove classes from tests (eggstorage)
Browse files Browse the repository at this point in the history
  • Loading branch information
jpmckinney committed Jul 19, 2024
1 parent 20f85a2 commit d86b525
Showing 1 changed file with 93 additions and 88 deletions.
181 changes: 93 additions & 88 deletions tests/test_eggstorage.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@
from unittest.mock import patch

import pytest
from twisted.trial import unittest
from zope.interface import implementer
from zope.interface.verify import verifyObject

Expand All @@ -13,18 +12,6 @@
from scrapyd.interfaces import IEggStorage


@pytest.mark.parametrize(
("versions", "expected"),
[
(["zzz", "b", "ddd", "a", "x"], ["a", "b", "ddd", "x", "zzz"]),
(["10", "1", "9"], ["1", "9", "10"]),
(["2.11", "2.01", "2.9"], ["2.01", "2.9", "2.11"]),
],
)
def test_sorted_versions(versions, expected):
assert sorted_versions(versions) == expected


@implementer(IEggStorage)
class SomeFakeEggStorage:
def __init__(self, config):
Expand All @@ -46,104 +33,122 @@ def delete(self, project, version=None):
pass


class TestConfigureEggStorage(unittest.TestCase):
def test_egg_config_application(self):
config = Config()
eggstore = "tests.test_eggstorage.SomeFakeEggStorage"
config.cp.set("scrapyd", "eggstorage", eggstore)
app = application(config)
app_eggstorage = app.getComponent(IEggStorage)
@pytest.fixture()
def eggstorage(tmpdir):
return FilesystemEggStorage(Config(values={"eggs_dir": tmpdir}))


@pytest.mark.parametrize(
("versions", "expected"),
[
(["zzz", "b", "ddd", "a", "x"], ["a", "b", "ddd", "x", "zzz"]),
(["10", "1", "9"], ["1", "9", "10"]),
(["2.11", "2.01", "2.9"], ["2.01", "2.9", "2.11"]),
],
)
def test_sorted_versions(versions, expected):
assert sorted_versions(versions) == expected


def test_egg_config_application():
config = Config()
eggstore = "tests.test_eggstorage.SomeFakeEggStorage"
config.cp.set("scrapyd", "eggstorage", eggstore)
app = application(config)
app_eggstorage = app.getComponent(IEggStorage)

assert isinstance(app_eggstorage, SomeFakeEggStorage)
assert app_eggstorage.list_projects() == ["hello_world"]


def test_interface(eggstorage):
verifyObject(IEggStorage, eggstorage)


def test_put_secure(eggstorage):
with pytest.raises(DirectoryTraversalError) as exc:
eggstorage.put(BytesIO(b"egg01"), "../p", "v") # version is sanitized

assert str(exc.value) == "../p"


self.assertIsInstance(app_eggstorage, SomeFakeEggStorage)
self.assertEqual(app_eggstorage.list_projects(), ["hello_world"])
def test_get_secure(eggstorage):
with pytest.raises(DirectoryTraversalError) as exc:
eggstorage.get("../p", "v") # version is sanitized

assert str(exc.value) == "../p"

class EggStorageTest(unittest.TestCase):
def setUp(self):
d = self.mktemp()
config = Config(values={"eggs_dir": d})
self.eggst = FilesystemEggStorage(config)

def test_interface(self):
verifyObject(IEggStorage, self.eggst)
def test_list_secure_join(eggstorage):
with pytest.raises(DirectoryTraversalError) as exc:
eggstorage.list("../p")

def test_put_secure(self):
with pytest.raises(DirectoryTraversalError) as exc:
self.eggst.put(BytesIO(b"egg01"), "../p", "v") # version is sanitized
assert str(exc.value) == "../p"

self.assertEqual(str(exc.value), "../p")

def test_get_secure(self):
with pytest.raises(DirectoryTraversalError) as exc:
self.eggst.get("../p", "v") # version is sanitized
def test_list_secure_glob(eggstorage):
eggstorage.put(BytesIO(b"egg01"), "mybot", "01")
versions = eggstorage.list("*")

self.assertEqual(str(exc.value), "../p")
eggstorage.delete("mybot")
assert versions == [] # ['01'] if * not escaped

def test_list_secure_join(self):
with pytest.raises(DirectoryTraversalError) as exc:
self.eggst.list("../p")

self.assertEqual(str(exc.value), "../p")
def test_delete(eggstorage):
with pytest.raises(DirectoryTraversalError) as exc:
eggstorage.delete("../p", "v") # version is sanitized

def test_list_secure_glob(self):
self.eggst.put(BytesIO(b"egg01"), "mybot", "01")
versions = self.eggst.list("*")
assert str(exc.value) == "../p"

self.eggst.delete("mybot")
self.assertEqual(versions, []) # ['01'] if * not escaped

def test_delete(self):
with pytest.raises(DirectoryTraversalError) as exc:
self.eggst.delete("../p", "v") # version is sanitized
@patch("scrapyd.eggstorage.glob", new=lambda x: ["ddd", "abc", "bcaa"])
def test_list_hashes(eggstorage):
versions = eggstorage.list("any")

self.assertEqual(str(exc.value), "../p")
assert versions == ["abc", "bcaa", "ddd"]

@patch("scrapyd.eggstorage.glob", new=lambda x: ["ddd", "abc", "bcaa"])
def test_list_hashes(self):
versions = self.eggst.list("any")

self.assertEqual(versions, ["abc", "bcaa", "ddd"])
@patch("scrapyd.eggstorage.glob", new=lambda x: ["9", "2", "200", "3", "4"])
def test_list_semantic_versions(eggstorage):
versions = eggstorage.list("any")

@patch("scrapyd.eggstorage.glob", new=lambda x: ["9", "2", "200", "3", "4"])
def test_list_semantic_versions(self):
versions = self.eggst.list("any")
assert versions == ["2", "3", "4", "9", "200"]

self.assertEqual(versions, ["2", "3", "4", "9", "200"])

def test_put_get_list_delete(self):
self.eggst.put(BytesIO(b"egg01"), "mybot", "01")
self.eggst.put(BytesIO(b"egg03"), "mybot", "03/ver")
self.eggst.put(BytesIO(b"egg02"), "mybot", "02_my branch")
def test_put_get_list_delete(eggstorage):
eggstorage.put(BytesIO(b"egg01"), "mybot", "01")
eggstorage.put(BytesIO(b"egg03"), "mybot", "03/ver")
eggstorage.put(BytesIO(b"egg02"), "mybot", "02_my branch")

self.assertEqual(self.eggst.list("mybot"), ["01", "02_my_branch", "03_ver"])
self.assertEqual(self.eggst.list("mybot2"), [])
assert eggstorage.list("mybot") == ["01", "02_my_branch", "03_ver"]
assert eggstorage.list("mybot2") == []

v, f = self.eggst.get("mybot")
try:
self.assertEqual(v, "03_ver")
self.assertEqual(f.read(), b"egg03")
finally:
f.close()
v, f = eggstorage.get("mybot")
try:
assert v == "03_ver"
assert f.read() == b"egg03"
finally:
f.close()

v, f = self.eggst.get("mybot", "02_my branch")
try:
self.assertEqual(v, "02_my branch")
self.assertEqual(f.read(), b"egg02")
finally:
f.close()
v, f = eggstorage.get("mybot", "02_my branch")
try:
assert v == "02_my branch"
assert f.read() == b"egg02"
finally:
f.close()

v, f = self.eggst.get("mybot", "02_my_branch")
try:
self.assertEqual(v, "02_my_branch")
self.assertEqual(f.read(), b"egg02")
finally:
f.close()
v, f = eggstorage.get("mybot", "02_my_branch")
try:
assert v == "02_my_branch"
assert f.read() == b"egg02"
finally:
f.close()

self.eggst.delete("mybot", "02_my branch")
self.assertEqual(self.eggst.list("mybot"), ["01", "03_ver"])
eggstorage.delete("mybot", "02_my branch")
assert eggstorage.list("mybot") == ["01", "03_ver"]

self.eggst.delete("mybot", "03_ver")
self.assertEqual(self.eggst.list("mybot"), ["01"])
eggstorage.delete("mybot", "03_ver")
assert eggstorage.list("mybot") == ["01"]

self.eggst.delete("mybot")
self.assertEqual(self.eggst.list("mybot"), [])
eggstorage.delete("mybot")
assert eggstorage.list("mybot") == []

0 comments on commit d86b525

Please sign in to comment.