ci: Use black instead of flake8

It results in mostly the same style, plus it’s configurable via pyproject.toml.
This commit is contained in:
2023-11-01 05:04:43 +01:00
parent 50442c56fe
commit d1a0510734
15 changed files with 473 additions and 256 deletions

View File

@@ -47,115 +47,167 @@ SERIALIZED_M1 = b"""{
@pytest.fixture()
def local_feed():
secret = b64decode('Mz2qkNOP2K6upnqibWrR+z8pVUI1ReA1MLc7QMtF2qQ=')
secret = b64decode("Mz2qkNOP2K6upnqibWrR+z8pVUI1ReA1MLc7QMtF2qQ=")
return LocalFeed(SigningKey(secret))
@pytest.fixture()
def remote_feed():
public = b64decode('I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=')
public = b64decode("I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=")
return Feed(VerifyKey(public))
def test_local_feed():
secret = b64decode('Mz2qkNOP2K6upnqibWrR+z8pVUI1ReA1MLc7QMtF2qQ=')
secret = b64decode("Mz2qkNOP2K6upnqibWrR+z8pVUI1ReA1MLc7QMtF2qQ=")
feed = LocalFeed(SigningKey(secret))
assert bytes(feed.private_key) == secret
assert bytes(feed.public_key) == b64decode('I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=')
assert feed.id == '@I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=.ed25519'
assert bytes(feed.public_key) == b64decode("I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=")
assert feed.id == "@I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=.ed25519"
def test_remote_feed():
public = b64decode('I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=')
public = b64decode("I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=")
feed = Feed(VerifyKey(public))
assert bytes(feed.public_key) == public
assert feed.id == '@I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=.ed25519'
assert feed.id == "@I/4cyN/jPBbDsikbHzAEvmaYlaJK33lW3UhWjNXjyrU=.ed25519"
m1 = Message(feed, OrderedDict([
('type', 'about'),
('about', feed.id),
('name', 'neo'),
('description', 'The Chosen One')
]), 'foo', timestamp=1495706260190)
m1 = Message(
feed,
OrderedDict(
[
("type", "about"),
("about", feed.id),
("name", "neo"),
("description", "The Chosen One"),
]
),
"foo",
timestamp=1495706260190,
)
with pytest.raises(NoPrivateKeyException):
feed.sign(m1)
def test_local_message(local_feed):
m1 = LocalMessage(local_feed, OrderedDict([
('type', 'about'),
('about', local_feed.id),
('name', 'neo'),
('description', 'The Chosen One')
]), timestamp=1495706260190)
m1 = LocalMessage(
local_feed,
OrderedDict(
[
("type", "about"),
("about", local_feed.id),
("name", "neo"),
("description", "The Chosen One"),
]
),
timestamp=1495706260190,
)
assert m1.timestamp == 1495706260190
assert m1.previous is None
assert m1.sequence == 1
assert m1.signature == \
'lPsQ9P10OgeyH6u0unFgiI2wV/RQ7Q2x2ebxnXYCzsJ055TBMXphRADTKhOMS2EkUxXQ9k3amj5fnWPudGxwBQ==.sig.ed25519'
assert m1.key == '%xRDqws/TrQmOd4aEwZ32jdLhP873ZKjIgHlggPR0eoo=.sha256'
assert (
m1.signature
== "lPsQ9P10OgeyH6u0unFgiI2wV/RQ7Q2x2ebxnXYCzsJ055TBMXphRADTKhOMS2EkUxXQ9k3amj5fnWPudGxwBQ==.sig.ed25519"
)
assert m1.key == "%xRDqws/TrQmOd4aEwZ32jdLhP873ZKjIgHlggPR0eoo=.sha256"
m2 = LocalMessage(local_feed, OrderedDict([
('type', 'about'),
('about', local_feed.id),
('name', 'morpheus'),
('description', 'Dude with big jaw')
]), previous=m1, timestamp=1495706447426)
m2 = LocalMessage(
local_feed,
OrderedDict(
[
("type", "about"),
("about", local_feed.id),
("name", "morpheus"),
("description", "Dude with big jaw"),
]
),
previous=m1,
timestamp=1495706447426,
)
assert m2.timestamp == 1495706447426
assert m2.previous is m1
assert m2.sequence == 2
assert m2.signature == \
'3SY85LX6/ppOfP4SbfwZbKfd6DccbLRiB13pwpzbSK0nU52OEJxOqcJ2Uensr6RkrWztWLIq90sNOn1zRAoOAw==.sig.ed25519'
assert m2.key == '%nx13uks5GUwuKJC49PfYGMS/1pgGTtwwdWT7kbVaroM=.sha256'
assert (
m2.signature
== "3SY85LX6/ppOfP4SbfwZbKfd6DccbLRiB13pwpzbSK0nU52OEJxOqcJ2Uensr6RkrWztWLIq90sNOn1zRAoOAw==.sig.ed25519"
)
assert m2.key == "%nx13uks5GUwuKJC49PfYGMS/1pgGTtwwdWT7kbVaroM=.sha256"
def test_remote_message(remote_feed):
signature = 'lPsQ9P10OgeyH6u0unFgiI2wV/RQ7Q2x2ebxnXYCzsJ055TBMXphRADTKhOMS2EkUxXQ9k3amj5fnWPudGxwBQ==.sig.ed25519'
m1 = Message(remote_feed, OrderedDict([
('type', 'about'),
('about', remote_feed.id),
('name', 'neo'),
('description', 'The Chosen One')
]), signature, timestamp=1495706260190)
signature = "lPsQ9P10OgeyH6u0unFgiI2wV/RQ7Q2x2ebxnXYCzsJ055TBMXphRADTKhOMS2EkUxXQ9k3amj5fnWPudGxwBQ==.sig.ed25519"
m1 = Message(
remote_feed,
OrderedDict(
[
("type", "about"),
("about", remote_feed.id),
("name", "neo"),
("description", "The Chosen One"),
]
),
signature,
timestamp=1495706260190,
)
assert m1.timestamp == 1495706260190
assert m1.previous is None
assert m1.sequence == 1
assert m1.signature == signature
assert m1.key == '%xRDqws/TrQmOd4aEwZ32jdLhP873ZKjIgHlggPR0eoo=.sha256'
assert m1.key == "%xRDqws/TrQmOd4aEwZ32jdLhP873ZKjIgHlggPR0eoo=.sha256"
signature = '3SY85LX6/ppOfP4SbfwZbKfd6DccbLRiB13pwpzbSK0nU52OEJxOqcJ2Uensr6RkrWztWLIq90sNOn1zRAoOAw==.sig.ed25519'
m2 = Message(remote_feed, OrderedDict([
('type', 'about'),
('about', remote_feed.id),
('name', 'morpheus'),
('description', 'Dude with big jaw')
]), signature, previous=m1, timestamp=1495706447426)
signature = "3SY85LX6/ppOfP4SbfwZbKfd6DccbLRiB13pwpzbSK0nU52OEJxOqcJ2Uensr6RkrWztWLIq90sNOn1zRAoOAw==.sig.ed25519"
m2 = Message(
remote_feed,
OrderedDict(
[
("type", "about"),
("about", remote_feed.id),
("name", "morpheus"),
("description", "Dude with big jaw"),
]
),
signature,
previous=m1,
timestamp=1495706447426,
)
assert m2.timestamp == 1495706447426
assert m2.previous is m1
assert m2.sequence == 2
assert m2.signature == signature
m2.verify(signature)
assert m2.key == '%nx13uks5GUwuKJC49PfYGMS/1pgGTtwwdWT7kbVaroM=.sha256'
assert m2.key == "%nx13uks5GUwuKJC49PfYGMS/1pgGTtwwdWT7kbVaroM=.sha256"
def test_remote_no_signature(remote_feed):
with pytest.raises(ValueError):
Message(remote_feed, OrderedDict([
('type', 'about'),
('about', remote_feed.id),
('name', 'neo'),
('description', 'The Chosen One')
]), None, timestamp=1495706260190)
Message(
remote_feed,
OrderedDict(
[
("type", "about"),
("about", remote_feed.id),
("name", "neo"),
("description", "The Chosen One"),
]
),
None,
timestamp=1495706260190,
)
def test_serialize(local_feed):
m1 = LocalMessage(local_feed, OrderedDict([
('type', 'about'),
('about', local_feed.id),
('name', 'neo'),
('description', 'The Chosen One')
]), timestamp=1495706260190)
m1 = LocalMessage(
local_feed,
OrderedDict(
[
("type", "about"),
("about", local_feed.id),
("name", "neo"),
("description", "The Chosen One"),
]
),
timestamp=1495706260190,
)
assert m1.serialize() == SERIALIZED_M1
@@ -163,9 +215,9 @@ def test_serialize(local_feed):
def test_parse(local_feed):
m1 = LocalMessage.parse(SERIALIZED_M1, local_feed)
assert m1.content == {
'type': 'about',
'about': local_feed.id,
'name': 'neo',
'description': 'The Chosen One'
"type": "about",
"about": local_feed.id,
"name": "neo",
"description": "The Chosen One",
}
assert m1.timestamp == 1495706260190

View File

@@ -36,20 +36,25 @@ async def _collect_messages(generator):
results.append(msg)
return results
MSG_BODY_1 = (b'{"previous":"%KTGP6W8vF80McRAZHYDWuKOD0KlNyKSq6Gb42iuV7Iw=.sha256","author":"@1+Iwm79DKvVBqYKFkhT6fWRbA'
b'VvNNVH4F2BSxwhYmx8=.ed25519","sequence":116,"timestamp":1496696699331,"hash":"sha256","content":{"type"'
b':"post","channel":"crypto","text":"Does anybody know any good resources (e.g. books) to learn cryptogra'
b'phy? I\'m not speaking of basic concepts (e.g. what\'s a private key) but the actual mathematics behind'
b' the whole thing.\\nI have a copy of the \\"Handbook of Applied Cryptography\\" on my bookshelf but I f'
b'ound it too long/hard to follow. Are there any better alternatives?","mentions":[]},"signature":"hqKePb'
b'bTXWxEi1njDnOWFsL0M0AoNoWyBFgNE6KXj//DThepaZSy9vRbygDHX5uNmCdyOrsQrwZsZhmUYKwtDQ==.sig.ed25519"}')
MSG_BODY_2 = (b'{"previous":"%iQRhPyqmNLpGaO1Tpm1I22jqnUEwRwkCTDbwAGtM+lY=.sha256","author":"@1+Iwm79DKvVBqYKFkhT6fWRbA'
b'VvNNVH4F2BSxwhYmx8=.ed25519","sequence":103,"timestamp":1496674211806,"hash":"sha256","content":{"type"'
b':"post","channel":"git-ssb","text":"Is it only me or `git.scuttlebot.io` is timing out?\\n\\nE.g. try a'
b'ccessing %vZCTqraoqKBKNZeATErXEtnoEr+wnT3p8tT+vL+29I4=.sha256","mentions":[{"link":"%vZCTqraoqKBKNZeATE'
b'rXEtnoEr+wnT3p8tT+vL+29I4=.sha256"}]},"signature":"+i4U0HUGDDEyNoNr2NIROPnT3WQj3RuTaIhY5koWW8f0vwr4tZsY'
b'mAkqqMwFWfP+eBIbc7DZ835er6r6h9CwAg==.sig.ed25519"}')
MSG_BODY_1 = (
b'{"previous":"%KTGP6W8vF80McRAZHYDWuKOD0KlNyKSq6Gb42iuV7Iw=.sha256","author":"@1+Iwm79DKvVBqYKFkhT6fWRbA'
b'VvNNVH4F2BSxwhYmx8=.ed25519","sequence":116,"timestamp":1496696699331,"hash":"sha256","content":{"type"'
b':"post","channel":"crypto","text":"Does anybody know any good resources (e.g. books) to learn cryptogra'
b"phy? I'm not speaking of basic concepts (e.g. what's a private key) but the actual mathematics behind"
b' the whole thing.\\nI have a copy of the \\"Handbook of Applied Cryptography\\" on my bookshelf but I f'
b'ound it too long/hard to follow. Are there any better alternatives?","mentions":[]},"signature":"hqKePb'
b'bTXWxEi1njDnOWFsL0M0AoNoWyBFgNE6KXj//DThepaZSy9vRbygDHX5uNmCdyOrsQrwZsZhmUYKwtDQ==.sig.ed25519"}'
)
MSG_BODY_2 = (
b'{"previous":"%iQRhPyqmNLpGaO1Tpm1I22jqnUEwRwkCTDbwAGtM+lY=.sha256","author":"@1+Iwm79DKvVBqYKFkhT6fWRbA'
b'VvNNVH4F2BSxwhYmx8=.ed25519","sequence":103,"timestamp":1496674211806,"hash":"sha256","content":{"type"'
b':"post","channel":"git-ssb","text":"Is it only me or `git.scuttlebot.io` is timing out?\\n\\nE.g. try a'
b'ccessing %vZCTqraoqKBKNZeATErXEtnoEr+wnT3p8tT+vL+29I4=.sha256","mentions":[{"link":"%vZCTqraoqKBKNZeATE'
b'rXEtnoEr+wnT3p8tT+vL+29I4=.sha256"}]},"signature":"+i4U0HUGDDEyNoNr2NIROPnT3WQj3RuTaIhY5koWW8f0vwr4tZsY'
b'mAkqqMwFWfP+eBIbc7DZ835er6r6h9CwAg==.sig.ed25519"}'
)
class MockSHSSocket(SHSDuplexStream):
@@ -129,26 +134,28 @@ async def test_message_decoding(ps_client):
assert ps.is_connected
ps_client.feed([
b'\n\x00\x00\x00\x9a\x00\x00\x04\xfb',
b'{"name":["createHistoryStream"],"args":[{"id":"@omgyp7Pnrw+Qm0I6T6Fh5VvnKmodMXwnxTIesW2DgMg=.ed25519",'
b'"seq":10,"live":true,"keys":false}],"type":"source"}'
])
ps_client.feed(
[
b"\n\x00\x00\x00\x9a\x00\x00\x04\xfb",
b'{"name":["createHistoryStream"],"args":[{"id":"@omgyp7Pnrw+Qm0I6T6Fh5VvnKmodMXwnxTIesW2DgMg=.ed25519",'
b'"seq":10,"live":true,"keys":false}],"type":"source"}',
]
)
messages = (await _collect_messages(ps))
messages = await _collect_messages(ps)
assert len(messages) == 1
assert messages[0].type == PSMessageType.JSON
assert messages[0].body == {
'name': ['createHistoryStream'],
'args': [
"name": ["createHistoryStream"],
"args": [
{
'id': '@omgyp7Pnrw+Qm0I6T6Fh5VvnKmodMXwnxTIesW2DgMg=.ed25519',
'seq': 10,
'live': True,
'keys': False
"id": "@omgyp7Pnrw+Qm0I6T6Fh5VvnKmodMXwnxTIesW2DgMg=.ed25519",
"seq": 10,
"live": True,
"keys": False,
}
],
'type': 'source'
"type": "source",
}
@@ -160,26 +167,36 @@ async def test_message_encoding(ps_client):
assert ps.is_connected
ps.send({
'name': ['createHistoryStream'],
'args': [{
'id': "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
'seq': 1,
'live': False,
'keys': False
}],
'type': 'source'
}, stream=True)
ps.send(
{
"name": ["createHistoryStream"],
"args": [
{
"id": "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
"seq": 1,
"live": False,
"keys": False,
}
],
"type": "source",
},
stream=True,
)
header, body = list(ps_client.get_output())
assert header == b'\x0a\x00\x00\x00\xa6\x00\x00\x00\x01'
assert json.loads(body.decode('utf-8')) == {
assert header == b"\x0a\x00\x00\x00\xa6\x00\x00\x00\x01"
assert json.loads(body.decode("utf-8")) == {
"name": ["createHistoryStream"],
"args": [
{"id": "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519", "seq": 1, "live": False, "keys": False}
{
"id": "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
"seq": 1,
"live": False,
"keys": False,
}
],
"type": "source"
"type": "source",
}
@@ -188,56 +205,72 @@ async def test_message_stream(ps_client, mocker):
await ps_client.connect()
ps = PacketStream(ps_client)
mocker.patch.object(ps, 'register_handler', wraps=ps.register_handler)
mocker.patch.object(ps, "register_handler", wraps=ps.register_handler)
assert ps.is_connected
ps.send({
'name': ['createHistoryStream'],
'args': [{
'id': "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
'seq': 1,
'live': False,
'keys': False
}],
'type': 'source'
}, stream=True)
ps.send(
{
"name": ["createHistoryStream"],
"args": [
{
"id": "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
"seq": 1,
"live": False,
"keys": False,
}
],
"type": "source",
},
stream=True,
)
assert ps.req_counter == 2
assert ps.register_handler.call_count == 1
handler = list(ps._event_map.values())[0][1]
mock_process = mocker.AsyncMock()
mocker.patch.object(handler, 'process', mock_process)
mocker.patch.object(handler, "process", mock_process)
ps_client.feed([b'\n\x00\x00\x02\xc5\xff\xff\xff\xff', MSG_BODY_1])
ps_client.feed([b"\n\x00\x00\x02\xc5\xff\xff\xff\xff", MSG_BODY_1])
msg = await ps.read()
assert mock_process.await_count == 1
# responses have negative req
assert msg.req == -1
assert msg.body['previous'] == '%KTGP6W8vF80McRAZHYDWuKOD0KlNyKSq6Gb42iuV7Iw=.sha256'
assert msg.body["previous"] == "%KTGP6W8vF80McRAZHYDWuKOD0KlNyKSq6Gb42iuV7Iw=.sha256"
assert ps.req_counter == 2
stream_handler = ps.send({
'name': ['createHistoryStream'],
'args': [{
'id': "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
'seq': 1,
'live': False,
'keys': False
}],
'type': 'source'
}, stream=True)
stream_handler = ps.send(
{
"name": ["createHistoryStream"],
"args": [
{
"id": "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519",
"seq": 1,
"live": False,
"keys": False,
}
],
"type": "source",
},
stream=True,
)
assert ps.req_counter == 3
assert ps.register_handler.call_count == 2
handler = list(ps._event_map.values())[1][1]
mock_process = mocker.patch.object(handler, 'process', wraps=handler.process)
ps_client.feed([b'\n\x00\x00\x02\xc5\xff\xff\xff\xfe', MSG_BODY_1,
b'\x0e\x00\x00\x023\xff\xff\xff\xfe', MSG_BODY_2])
mock_process = mocker.patch.object(handler, "process", wraps=handler.process)
ps_client.feed(
[
b"\n\x00\x00\x02\xc5\xff\xff\xff\xfe",
MSG_BODY_1,
b"\x0e\x00\x00\x023\xff\xff\xff\xfe",
MSG_BODY_2,
]
)
# execute both message polling and response handling loops
collected, handled = await gather(_collect_messages(ps), _collect_messages(stream_handler))
@@ -258,29 +291,30 @@ async def test_message_request(ps_server, mocker):
ps = PacketStream(ps_server)
mocker.patch.object(ps, 'register_handler', wraps=ps.register_handler)
mocker.patch.object(ps, "register_handler", wraps=ps.register_handler)
ps.send({
'name': ['whoami'],
'args': []
})
ps.send({"name": ["whoami"], "args": []})
header, body = list(ps_server.get_output())
assert header == b'\x02\x00\x00\x00 \x00\x00\x00\x01'
assert json.loads(body.decode('utf-8')) == {"name": ["whoami"], "args": []}
assert header == b"\x02\x00\x00\x00 \x00\x00\x00\x01"
assert json.loads(body.decode("utf-8")) == {"name": ["whoami"], "args": []}
assert ps.req_counter == 2
assert ps.register_handler.call_count == 1
handler = list(ps._event_map.values())[0][1]
mock_process = mocker.AsyncMock()
mocker.patch.object(handler, 'process', mock_process)
ps_server.feed([b'\x02\x00\x00\x00>\xff\xff\xff\xff',
b'{"id":"@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519"}'])
mocker.patch.object(handler, "process", mock_process)
ps_server.feed(
[
b"\x02\x00\x00\x00>\xff\xff\xff\xff",
b'{"id":"@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519"}',
]
)
msg = await ps.read()
assert mock_process.await_count == 1
# responses have negative req
assert msg.req == -1
assert msg.body['id'] == '@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519'
assert msg.body["id"] == "@1+Iwm79DKvVBqYKFkhT6fWRbAVvNNVH4F2BSxwhYmx8=.ed25519"
assert ps.req_counter == 2

View File

@@ -38,21 +38,21 @@ CONFIG_FILE = """
}
"""
CONFIG_FILE_INVALID = CONFIG_FILE.replace('ed25519', 'foo')
CONFIG_FILE_INVALID = CONFIG_FILE.replace("ed25519", "foo")
def test_load_secret():
with patch('ssb.util.open', mock_open(read_data=CONFIG_FILE), create=True):
with patch("ssb.util.open", mock_open(read_data=CONFIG_FILE), create=True):
secret = load_ssb_secret()
priv_key = b'\xfd\xba\x83\x04\x8f\xef\x18\xb0\xf9\xab-\xc6\xc4\xcb \x1cX\x18"\xba\xd8\xd3\xc2_O5\x1a\t\x84\xfa\xc7A'
assert secret['id'] == '@rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=.ed25519'
assert bytes(secret['keypair']) == priv_key
assert bytes(secret['keypair'].verify_key) == b64decode('rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=')
assert secret["id"] == "@rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=.ed25519"
assert bytes(secret["keypair"]) == priv_key
assert bytes(secret["keypair"].verify_key) == b64decode("rsYpBIcXsxjQAf0JNes+MHqT2DL+EfopWKAp4rGeEPQ=")
def test_load_exception():
with pytest.raises(ConfigException):
with patch('ssb.util.open', mock_open(read_data=CONFIG_FILE_INVALID), create=True):
with patch("ssb.util.open", mock_open(read_data=CONFIG_FILE_INVALID), create=True):
load_ssb_secret()