fix(lint): reformat & resolve discovered Ruff errors

[rmlibre]

Description

Quite a few issues were found when running Ruff on the codebase. The benign included unused imports, formatting style choices, & complaints related to import * clauses primarily in the __init__.py files of subpackages. The package's standard for styling is declared as black in the readme, but it's only loosely abiding by that formatting. An automated formatter to enforce style standards is missing from the project.

Moreover, there were some NameError bugs in branching code. These were the offending lines:

Bug 1: Issue was missing as an import

This resulted in NameError being raised instead of PermissionError.

aiootp/aiootp/databases/async_database.py

Line 654 in 5ca2d61

raise Issue.cant_reassign_attribute(tag) # fixed v0.23.9

aiootp/aiootp/databases/sync_database.py

Line 620 in 5ca2d61

raise Issue.cant_reassign_attribute(tag) # fixed v0.23.9

Bug 2: sleep was missing as an import

The impact is uncertain, as these lines of code weren't reached in tests. They were intended to be guard clauses against the potential of simultaneous data buffering. These clauses weren't sufficient solutions to concurrency race conditions.

aiootp/aiootp/ciphers/cipher_streams.py

Lines 459 to 460 in 5ca2d61

	while self._is_digesting: # TODO: race conditions?
	sleep(0.00001) # pragma: no cover

aiootp/aiootp/ciphers/decipher_streams.py

Lines 576 to 577 in 5ca2d61

	while self._is_digesting: # TODO: race conditions?
	sleep(0.00001) # pragma: no cover

Expected behavior

When the configured Ruff linter & formatter are run on a release:

• There are no reported errors

• There are no changes to formatting

• The result is the canonical style standard for the package, with the only exceptions being for changes that would harm readability, hinder the communication of essential information, or are oddities / inconsistencies that would otherwise merit the raising of an issue upstream.

Remediations

Fix Bug 1:

• Add missing import

Fix Bug 2:

• Add missing import

• Write proper concurrency protection

The critical sections were wrapped in the new ConcurrencyGuard class. It works by assigning each context a unique, random authorization token & appending it to a double-ended queue (collections.deque). Depending on the class, it (a)synchronously sleeps until its token is at the 0th index of the queue. At that time, the code block wrapped by the instance's context manager is allowed to run. When it's done, it calls popleft() on the queue to take its token out of the queue so the next instance can run. If the popleft() call returns a different token, then IncoherentConcurrencyState exception is raised, indicating a potentially corrupt control flow / buffering of data.

aiootp/aiootp/ciphers/cipher_streams.py

Lines 242 to 246 in c9c13c6

	async with ConcurrencyGuard(self._digesting_now):
	self._byte_count += len(data)
	data = io.BytesIO(data).read
	_buffer, append = self._buffer_shortcuts
	await self._adigest_data(data, _buffer, append)

aiootp/aiootp/ciphers/cipher_streams.py

Lines 455 to 459 in c9c13c6

	with ConcurrencyGuard(self._digesting_now, probe_delay=0.0001):
	self._byte_count += len(data)
	data = io.BytesIO(data).read
	_buffer, append = self._buffer_shortcuts
	self._digest_data(data, _buffer, append)

aiootp/aiootp/ciphers/decipher_streams.py

Lines 307 to 310 in c9c13c6

	async with ConcurrencyGuard(self._digesting_now):
	data = io.BytesIO(data).read
	atest_block_id, append = self._buffer_shortcuts
	await self._adigest_data(data, atest_block_id, append)

aiootp/aiootp/ciphers/decipher_streams.py

Lines 581 to 584 in c9c13c6

	with ConcurrencyGuard(self._digesting_now):
	data = io.BytesIO(data).read
	atest_block_id, append = self._buffer_shortcuts
	self._digest_data(data, atest_block_id, append)

That class is defined here:

aiootp/aiootp/asynchs/concurrency_interface.py

Lines 43 to 75 in c9c13c6

	class ConcurrencyGuard(OpenFrozenTypedSlots):
	"""
	An interface for queueing execution contexts given only a shared
	`deque` or `deque`-like double-ended queue. Prevents simultaneous /
	out of order runs of blocks of code. A `deque` is recommended since
	it supports atomic operations. Any atomic, shared datastructure with
	`append`, `popleft`, & queue[0] methods would fit the API.
	_____________________________________
	| |
	| Example As Diagram: |
	|_____________________________________|
	----------------------
	| Shared Context |
	----------------------
	queue = deque()
	-----------------------
	----------------- | -----------------
	| Context A | | | Context B |
	----------------- | -----------------
	|
	with ConcurrencyGuard(queue): | with ConcurrencyGuard(queue):
	mutable_thing[0] = 0 | assert mutable_thing[0] == "done."
	... | mutable_thing[0] = 1
	... |
	assert mutable_thing[0] == 0 |
	mutable_thing[0] = "done." |
	|
	----------------------------------------------------------------------
	Context A is called first, but Context B is called before it finishes.
	----------------------------------------------------------------------

• Test the new implementation

aiootp/tests/test_online_cipher_interfaces.py

Lines 263 to 321 in c9c13c6

	async def test_async_concurrency_handling(self) -> None:
	config, cipher, salt, aad = choice(all_ciphers)
	chunk_size = 1024 * config.BLOCKSIZE
	data_a = (chunk_size * b"a")[
	config.INNER_HEADER_BYTES : -config.SENTINEL_BYTES
	]
	data_b = chunk_size * b"b"
	stream_enc = await cipher.astream_encrypt(salt=salt, aad=aad)
	fut_a = asynchs.new_task(stream_enc.abuffer(data_a))
	await asleep(0.00001)
	fut_b = asynchs.new_task(stream_enc.abuffer(data_b))
	await fut_a
	await fut_b
	ct = [b"".join(id_ct) async for id_ct in stream_enc.afinalize()]
	stream_dec = await cipher.astream_decrypt(
	salt=salt, aad=aad, iv=stream_enc.iv
	)
	ct_a = b"".join(ct[: len(ct) // 2])
	ct_b = b"".join(ct[len(ct) // 2 :])
	fut_a = asynchs.new_task(stream_dec.abuffer(ct_a))
	await asleep(0.00001)
	fut_b = asynchs.new_task(stream_dec.abuffer(ct_b))
	await fut_a
	await fut_b
	pt = b"".join([pt async for pt in stream_dec.afinalize()])
	assert data_a + data_b == pt
	async def test_sync_concurrency_handling(self) -> None:
	config, cipher, salt, aad = choice(all_ciphers)
	chunk_size = 16 * 1024 * config.BLOCKSIZE
	data_a = (chunk_size * b"a")[
	config.INNER_HEADER_BYTES : -config.SENTINEL_BYTES
	]
	data_b = chunk_size * b"b"
	stream_enc = cipher.stream_encrypt(salt=salt, aad=aad)
	fut_a = Threads.submit(stream_enc.buffer, data_a)
	asynchs.sleep(0.001)
	fut_b = Threads.submit(stream_enc.buffer, data_b)
	fut_a.result()
	fut_b.result()
	ct = [b"".join(id_ct) for id_ct in stream_enc.finalize()]
	stream_dec = cipher.stream_decrypt(
	salt=salt, aad=aad, iv=stream_enc.iv
	)
	ct_a = b"".join(ct[: len(ct) // 2])
	ct_b = b"".join(ct[len(ct) // 2 :])
	fut_a = Threads.submit(stream_dec.buffer, ct_a)
	asynchs.sleep(0.001)
	fut_b = Threads.submit(stream_dec.buffer, ct_b)
	fut_a.result()
	fut_b.result()
	pt = b"".join(stream_dec.finalize())
	assert data_a + data_b == pt

aiootp/tests/test_ConcurrencyGuard.py

Lines 21 to 64 in c9c13c6

	class TestConcurrencyGuard:
	async def test_detects_async_out_of_order_execution(self) -> None:
	problem = ( # fmt: skip
	"Another execution authorization token was allowed to skip "
	"the current's place in line."
	)
	error = ConcurrencyGuard.IncoherentConcurrencyState
	with Ignore(error, if_else=violation(problem)):
	queue = deque()
	async with ConcurrencyGuard(queue):
	queue.appendleft(token_bytes(32))
	with Ignore(error, if_else=violation(problem)):
	queue = deque()
	async with ConcurrencyGuard(queue):
	queue[0] = token_bytes(32)
	with Ignore(IndexError, if_else=violation(problem)):
	queue = deque()
	async with ConcurrencyGuard(queue):
	queue.popleft()
	async def test_detects_sync_out_of_order_execution(self) -> None:
	problem = ( # fmt: skip
	"Another execution authorization token was allowed to skip "
	"the current's place in line."
	)
	error = ConcurrencyGuard.IncoherentConcurrencyState
	with Ignore(error, if_else=violation(problem)):
	queue = deque()
	with ConcurrencyGuard(queue):
	queue.appendleft(token_bytes(32))
	with Ignore(error, if_else=violation(problem)):
	queue = deque()
	with ConcurrencyGuard(queue):
	queue[0] = token_bytes(32)
	with Ignore(IndexError, if_else=violation(problem)):
	queue = deque()
	with ConcurrencyGuard(queue):
	queue.popleft()

High-level style choices:

• Make style choices to become standard for the package.

The options for Ruff have been configured & annotated with rationale. For now, what Ruff says when running in that configuration defines most of the style guide. The caveats are described in this PR, but will be subject to refinement in the future. The work of fully documenting all style choices may not be necessary, albeit desirable. However, that'll be the work of a future PR.

• Automate style choices by implementing a ruff.toml or pyproject.toml.

aiootp/ruff.toml

Lines 37 to 48 in f9cf094

	[lint]
	select = ["B", "C", "E", "F", "Q", "W"]
	ignore = [
	"B006", # mutable defaults only used for **unpacking interface
	"B008", # random defaults at startup are intended
	"B028", # thanks for the warning, if stacklevel was needed
	"C408", # string keys aren't better than keyword arguments
	"E501", # trust the 76-length enforcement from the formatter
	"E731", # TODO: look through lambda's to maybe comply
	]
	fixable = ["ALL"]
	unfixable = []

Add lint & format automation to GitHub workflows:

• Linux Build

aiootp/.github/workflows/linux-python-app.yml

Lines 23 to 25 in f9cf094

	- name: lint with ruff
	run: |
	ruff check . && ruff format --line-length=76 --check .

• Windows Build

aiootp/.github/workflows/windows-python-app.yml

Lines 23 to 25 in f9cf094

	- name: lint with ruff
	run: |
	ruff check . && ruff format --line-length=76 --check .

• Mac Build

aiootp/.github/workflows/macos-python-app.yml

Lines 23 to 25 in f9cf094

	- name: lint with ruff
	run: |
	ruff check . && ruff format --line-length=76 --check .