refactoring to fix a bunch of test warnings

.vscode/settings.json

···
 		"-p",
 		"test_*.py"
 	],
-	"python.testing.pytestEnabled": false,
-	"python.testing.unittestEnabled": true
+	"python.testing.pytestEnabled": true,
+	"python.testing.unittestEnabled": false,
+	"python.testing.pytestArgs": [
+		"."
+	]
 }

pyproject.toml

···
 indent-style = "tab"
 
 [tool.setuptools_scm]
+
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"

src/millipds/__main__.py

···
 from getpass import getpass
 
 from docopt import docopt
+import aiohttp
 
 import cbrrr
 
···
 		else:
 			print("invalid account subcommand")
 	elif args["run"]:
-		asyncio.run(
-			service.run(
-				db=db,
-				sock_path=args["--sock_path"],
-				host=args["--listen_host"],
-				port=int(args["--listen_port"]),
-			)
-		)
+		async def run_with_client():
+			async with aiohttp.ClientSession() as client:
+				await service.run(
+					db=db,
+					client=client,
+					sock_path=args["--sock_path"],
+					host=args["--listen_host"],
+					port=int(args["--listen_port"]),
+				)
+		asyncio.run(run_with_client())
 	else:
 		print("CLI arg parse error?!")
 

src/millipds/app_util.py

···
 
 from . import database
 
+MILLIPDS_DB = web.AppKey("MILLIPDS_DB", database.Database)
+MILLIPDS_AIOHTTP_CLIENT = web.AppKey("MILLIPDS_AIOHTTP_CLIENT", aiohttp.ClientSession)
+MILLIPDS_FIREHOSE_QUEUES = web.AppKey("MILLIPDS_FIREHOSE_QUEUES", Set[asyncio.Queue[Optional[Tuple[int, bytes]]]])
+MILLIPDS_FIREHOSE_QUEUES_LOCK = web.AppKey("MILLIPDS_FIREHOSE_QUEUES_LOCK", asyncio.Lock)
 
 # these helpers are useful for conciseness and type hinting
-def get_db(req: web.Request) -> database.Database:
-	return req.app["MILLIPDS_DB"]
+def get_db(req: web.Request):
+	return req.app[MILLIPDS_DB]
 
-def get_client(req: web.Request) -> aiohttp.ClientSession:
-	return req.app["MILLIPDS_AIOHTTP_CLIENT"]
+def get_client(req: web.Request):
+	return req.app[MILLIPDS_AIOHTTP_CLIENT]
 
-def get_firehose_queues(req: web.Request) -> Set[asyncio.Queue[Optional[Tuple[int, bytes]]]]:
-	return req.app["MILLIPDS_FIREHOSE_QUEUES"]
+def get_firehose_queues(req: web.Request):
+	return req.app[MILLIPDS_FIREHOSE_QUEUES]
 
-def get_firehose_queues_lock(req: web.Request) -> asyncio.Lock:
-	return req.app["MILLIPDS_FIREHOSE_QUEUES_LOCK"]
+def get_firehose_queues_lock(req: web.Request):
+	return req.app[MILLIPDS_FIREHOSE_QUEUES_LOCK]

src/millipds/atproto_repo.py

···
 @routes.get("/xrpc/com.atproto.repo.describeRepo")
 async def repo_describe_repo(request: web.Request):
 	if "repo" not in request.query:
-		return web.HTTPBadRequest(text="missing repo")
+		raise web.HTTPBadRequest(text="missing repo")
 	did_or_handle = request.query["repo"]
 	with get_db(request).new_con(readonly=True) as con:
 		user_id, did, handle = con.execute(
···
 @routes.get("/xrpc/com.atproto.repo.getRecord")
 async def repo_get_record(request: web.Request):
 	if "repo" not in request.query:
-		return web.HTTPBadRequest(text="missing repo")
+		raise web.HTTPBadRequest(text="missing repo")
 	if "collection" not in request.query:
-		return web.HTTPBadRequest(text="missing collection")
+		raise web.HTTPBadRequest(text="missing collection")
 	if "rkey" not in request.query:
-		return web.HTTPBadRequest(text="missing rkey")
+		raise web.HTTPBadRequest(text="missing rkey")
 	did_or_handle = request.query["repo"]
 	collection = request.query["collection"]
 	rkey = request.query["rkey"]
···
 	).fetchone()
 	if row is None:
 		return await service_proxy(request) # forward to appview
-		#return web.HTTPNotFound(text="record not found")
+		#raise web.HTTPNotFound(text="record not found")
 	cid_out, value = row
 	cid_out = cbrrr.CID(cid_out)
 	if cid_in is not None:
 		if cbrrr.CID.decode(cid_in) != cid_out:
-			return web.HTTPNotFound(text="record not found with matching CID")
+			raise web.HTTPNotFound(text="record not found with matching CID")
 	return web.json_response({
 		"uri": f"at://{did_or_handle}/{collection}/{rkey}", # TODO rejig query to get the did out always,
 		"cid": cid_out.encode(),
···
 @routes.get("/xrpc/com.atproto.repo.listRecords")
 async def repo_list_records(request: web.Request):
 	if "repo" not in request.query:
-		return web.HTTPBadRequest(text="missing repo")
+		raise web.HTTPBadRequest(text="missing repo")
 	if "collection" not in request.query:
-		return web.HTTPBadRequest(text="missing collection")
+		raise web.HTTPBadRequest(text="missing collection")
 	limit = int(request.query.get("limit", 50))
 	if limit < 1 or limit > 100:
-		return web.HTTPBadRequest(text="limit out of range")
+		raise web.HTTPBadRequest(text="limit out of range")
 	reverse = request.query.get("reverse") == "true"
 	cursor = request.query.get("cursor", "" if reverse else "\xff")
 	did_or_handle = request.query["repo"]

src/millipds/atproto_sync.py

···
 			(request.query["did"], bytes(cbrrr.CID.decode(request.query["cid"]))) # TODO: check params exist first, give nicer error
 		).fetchone()
 		if blob_id is None:
-			return web.HTTPNotFound(text="blob not found")
+			raise web.HTTPNotFound(text="blob not found")
 		res = web.StreamResponse(headers={"Content-Disposition": f'attachment; filename="{request.query["cid"]}.bin"'})
 		res.content_type = "application/octet-stream"
 		await res.prepare(request)
···
 async def sync_get_blocks(request: web.Request):
 	did = request.query.get("did")
 	if did is None:
-		return web.HTTPBadRequest(text="no did specified")
+		raise web.HTTPBadRequest(text="no did specified")
 	try:
 		cids = [bytes(cbrrr.CID.decode(cid)) for cid in request.query.getall("cids")]
 	except ValueError:
-		return web.HTTPBadRequest(text="invalid cid")
+		raise web.HTTPBadRequest(text="invalid cid")
 	db = get_db(request)
 	row = db.con.execute("SELECT id FROM user WHERE did=?", (did,)).fetchone()
 	if row is None:
-		return web.HTTPNotFound(text="did not found")
+		raise web.HTTPNotFound(text="did not found")
 	user_id = row[0]
 	res = web.StreamResponse()
 	res.content_type = "application/vnd.ipld.car"
···
 async def sync_get_latest_commit(request: web.Request):
 	did = request.query.get("did")
 	if did is None:
-		return web.HTTPBadRequest(text="no did specified")
+		raise web.HTTPBadRequest(text="no did specified")
 	row = get_db(request).con.execute(
 		"SELECT rev, head FROM user WHERE did=?",
 		(did,)
 	).fetchone()
 	if row is None:
-		return web.HTTPNotFound(text="did not found")
+		raise web.HTTPNotFound(text="did not found")
 	rev, head = row
 	return web.json_response({
 		"cid": cbrrr.CID(head).encode(),
···
 @routes.get("/xrpc/com.atproto.sync.getRecord")
 async def sync_get_record(request: web.Request):
 	if "did" not in request.query:
-		return web.HTTPBadRequest(text="missing did")
+		raise web.HTTPBadRequest(text="missing did")
 	if "collection" not in request.query:
-		return web.HTTPBadRequest(text="missing collection")
+		raise web.HTTPBadRequest(text="missing collection")
 	if "rkey" not in request.query:
-		return web.HTTPBadRequest(text="missing rkey")
+		raise web.HTTPBadRequest(text="missing rkey")
 
 	# we don't stream the response because it should be compact-ish
 	car = repo_ops.get_record(
···
 	)
 
 	if car is None:
-		return web.HTTPNotFound(text="did or record not found")
+		raise web.HTTPNotFound(text="did or record not found")
 	
 	return web.Response(
 		body=car,
···
 async def sync_get_repo_status(request: web.Request):
 	did = request.query.get("did")
 	if did is None:
-		return web.HTTPBadRequest(text="no did specified")
+		raise web.HTTPBadRequest(text="no did specified")
 	row = get_db(request).con.execute(
 		"SELECT rev FROM user WHERE did=?",
 		(did,)
 	).fetchone()
 	if row is None:
-		return web.HTTPNotFound(text="did not found")
+		raise web.HTTPNotFound(text="did not found")
 	return web.json_response({
 		"did": did,
 		"active": True,
···
 async def sync_get_repo(request: web.Request):
 	did = request.query.get("did")
 	if did is None:
-		return web.HTTPBadRequest(text="no did specified")
+		raise web.HTTPBadRequest(text="no did specified")
 	since = request.query.get("since", "") # empty string is "lowest" possible value wrt string comparison
 
 	# TODO: do bad things happen if a client holds the connection open for a long time?
···
 				(did,)
 			).fetchone()
 		except TypeError: # from trying to unpack None
-			return web.HTTPNotFound(text="repo not found")
+			raise web.HTTPNotFound(text="repo not found")
 
 		res = web.StreamResponse()
 		res.content_type = "application/vnd.ipld.car"
···
 async def sync_list_blobs(request: web.Request):
 	did = request.query.get("did")
 	if did is None:
-		return web.HTTPBadRequest(text="no did specified")
+		raise web.HTTPBadRequest(text="no did specified")
 	since = request.query.get("since", "") # empty string is "lowest" possible value wrt string comparison
 	limit = int(request.query.get("limit", 500))
 	if limit < 1 or limit > 1000:
-		return web.HTTPBadRequest(text="limit out of range")
+		raise web.HTTPBadRequest(text="limit out of range")
 	cursor = int(request.query.get("cursor", 0))
 
 	cids = []

src/millipds/auth_bearer.py

···
 
 routes = web.RouteTableDef()
 def authenticated(handler):
-	def authentication_handler(request: web.Request, *args, **kwargs):
+	async def authentication_handler(request: web.Request, *args, **kwargs):
 		# extract the auth token
 		auth = request.headers.get("Authorization")
 		if auth is None:
···
 				key=db.config["jwt_access_secret"],
 				algorithms=["HS256"],
 				audience=db.config["pds_did"],
-				require=["exp", "scope"],  # consider iat?
-				strict_aud=True,
+				options={
+					"require": ["exp", "iat", "scope"],  # consider iat?
+					"verify_exp": True,
+					"verify_iat": True,
+					"strict_aud": True, # may be unnecessary
+				}
 			)
 		except jwt.exceptions.PyJWTError:
 			raise web.HTTPUnauthorized(text="invalid jwt")
···
 		if not subject.startswith("did:"):
 			raise web.HTTPUnauthorized(text="invalid jwt: invalid subject")
 		request["authed_did"] = subject
-		return handler(request, *args, **kwargs)
+		return await handler(request, *args, **kwargs)
 
 	return authentication_handler

src/millipds/auth_oauth.py

···
 		# TODO: verify iat?, iss?
 
 		if request.method != decoded["htm"]:
-			return web.HTTPBadRequest(
+			raise web.HTTPBadRequest(
 				text="dpop: bad htm"
 			)
 
 		if str(request.url) != decoded["htu"]:
 			logger.info(f"{request.url!r} != {decoded['htu']!r}")
-			return web.HTTPBadRequest(
+			raise web.HTTPBadRequest(
 				text="dpop: bad htu (if your application is reverse-proxied, make sure the Host header is getting set properly)"
 			)
 		
 		if decoded.get("nonce") != DPOP_NONCE:
-			return web.HTTPBadRequest(
+			raise web.HTTPBadRequest(
 				body=json.dumps({
 					"error": "use_dpop_nonce",
 					"error_description": "Authorization server requires nonce in DPoP proof"

src/millipds/database.py

···
 				"INSERT INTO mst(repo, cid, since, value) VALUES (?, ?, ?, ?)",
 				(user_id, bytes(empty_mst.cid), tid, empty_mst.serialised),
 			)
-			#util.mkdirs_for_file(repo_path)
-			#UserDatabase.init_tables(self.con, did, repo_path, tid)
-		#self.con.execute("DETACH spoke")
 
 	def verify_account_login(
 		self, did_or_handle: str, password: str

src/millipds/service.py

···
 	# TODO: forward to appview(?) if we can't answer?
 	handle = request.query.get("handle")
 	if handle is None:
-		return web.HTTPBadRequest(text="missing or invalid handle")
+		raise web.HTTPBadRequest(text="missing or invalid handle")
 	did = get_db(request).did_by_handle(handle)
 	if not did:
-		return web.HTTPNotFound(text="no user by that handle exists on this PDS")
+		raise web.HTTPNotFound(text="no user by that handle exists on this PDS")
 	return web.json_response({"did": did})
 
 
···
 	try:
 		req_json: dict = await request.json()
 	except json.JSONDecodeError:
-		return web.HTTPBadRequest(text="expected JSON")
+		raise web.HTTPBadRequest(text="expected JSON")
 
 	identifier = req_json.get("identifier")
 	password = req_json.get("password")
 	if not (isinstance(identifier, str) and isinstance(password, str)):
-		return web.HTTPBadRequest(text="invalid identifier or password")
+		raise web.HTTPBadRequest(text="invalid identifier or password")
 
 	# do authentication
 	db = get_db(request)
···
 	req_json: dict = await request.json()
 	handle = req_json.get("handle")
 	if handle is None:
-		return web.HTTPBadRequest(text="missing or invalid handle")
+		raise web.HTTPBadRequest(text="missing or invalid handle")
 	# TODO: actually validate it, and update the db!!!
 	# (I'm writing this half-baked version just so I can send firehose #identity events)
 	with get_db(request).new_con() as con:
···
 	)
 
 
-def construct_app(routes, db: database.Database) -> web.Application:
+def construct_app(routes, db: database.Database, client: aiohttp.ClientSession) -> web.Application:
 	cors = cors_middleware( # TODO: review and reduce scope - and maybe just /xrpc/*?
 		allow_all=True,
 		expose_headers=["*"],
···
 		allow_credentials=True,
 		max_age=100_000_000
 	)
+
+	client.headers.update({"User-Agent": importlib.metadata.version("millipds")})
 
 	app = web.Application(middlewares=[cors, atproto_service_proxy_middleware])
-	app["MILLIPDS_DB"] = db
-	app["MILLIPDS_AIOHTTP_CLIENT"] = (
-		aiohttp.ClientSession()
-	)  # should this be dependency-injected?
-	app["MILLIPDS_FIREHOSE_QUEUES"] = set()
-	app["MILLIPDS_FIREHOSE_QUEUES_LOCK"] = asyncio.Lock()
+	app[MILLIPDS_DB] = db
+	app[MILLIPDS_AIOHTTP_CLIENT] = client
+	app[MILLIPDS_FIREHOSE_QUEUES] = set()
+	app[MILLIPDS_FIREHOSE_QUEUES_LOCK] = asyncio.Lock()
 	app.add_routes(routes)
 	app.add_routes(auth_oauth.routes)
 	app.add_routes(atproto_sync.routes)
···
 	return app
 
 
-async def run(db: database.Database, sock_path: Optional[str], host: str, port: int):
+async def run(db: database.Database, client: aiohttp.ClientSession, sock_path: Optional[str], host: str, port: int):
 	"""
 	This gets invoked via millipds.__main__.py
 	"""
 
-	app = construct_app(routes, db)
+	app = construct_app(routes, db, client)
 	runner = web.AppRunner(app, access_log_format=static_config.HTTP_LOG_FMT)
 	await runner.setup()
 

tests/integration_test.py

···
 import pytest
 import dataclasses
 import aiohttp
+import aiohttp.web
 
 from millipds import service
 from millipds import database
 from millipds import crypto
 
 @dataclasses.dataclass
-class TestPDS:
+class PDSInfo:
 	endpoint: str
 	db: database.Database
 
 old_web_tcpsite_start = aiohttp.web.TCPSite.start
 
-def make_capture_random_bound_port_web_tcpsite_startstart(queue):
-	async def mock_start(site, *args, **kwargs):
+def make_capture_random_bound_port_web_tcpsite_startstart(queue: asyncio.Queue):
+	async def mock_start(site: aiohttp.web.TCPSite, *args, **kwargs):
 		nonlocal queue
 		await old_web_tcpsite_start(site, *args, **kwargs)
 		await queue.put(site._server.sockets[0].getsockname()[1])
 	return mock_start
 
-async def service_run_and_capture_port(queue,  **kwargs):
+async def service_run_and_capture_port(queue: asyncio.Queue, **kwargs):
 	mock_start = make_capture_random_bound_port_web_tcpsite_startstart(queue)
 	with unittest.mock.patch.object(aiohttp.web.TCPSite, "start", new=mock_start):
 		await service.run(**kwargs)
···
 async def test_pds(aiolib):
 	queue = asyncio.Queue()
 	with tempfile.TemporaryDirectory() as tempdir:
-		db_path = f"{tempdir}/millipds-0000.db"
-		db = database.Database(path=db_path)
-
-		hostname = "localhost:0"
-		db.update_config(
-			pds_pfx=f'http://{hostname}',
-			pds_did=f'did:web:{urllib.parse.quote(hostname)}',
-			bsky_appview_pfx="https://api.bsky.app",
-			bsky_appview_did="did:web:api.bsky.app",
-		)
+		async with aiohttp.ClientSession() as client:
+			db_path = f"{tempdir}/millipds-0000.db"
+			db = database.Database(path=db_path)
 
-		service_run_task = asyncio.create_task(
-			service_run_and_capture_port(
-				queue,
-				db=db,
-				sock_path=None,
-				host="localhost",
-				port=0,
+			hostname = "localhost:0"
+			db.update_config(
+				pds_pfx=f'http://{hostname}',
+				pds_did=f'did:web:{urllib.parse.quote(hostname)}',
+				bsky_appview_pfx="https://api.bsky.app",
+				bsky_appview_did="did:web:api.bsky.app",
 			)
-		)
-		queue_get_task = asyncio.create_task(queue.get())
-		done, pending = await asyncio.wait(
-			(queue_get_task, service_run_task),
-			return_when=asyncio.FIRST_COMPLETED,
-		)
-		if done == service_run_task:
-			raise service_run_task.execption()
-		else:
-			port = queue_get_task.result()
 
-		hostname = f"localhost:{port}"
-		db.update_config(
-			pds_pfx=f'http://{hostname}',
-			pds_did=f'did:web:{urllib.parse.quote(hostname)}',
-			bsky_appview_pfx="https://api.bsky.app",
-			bsky_appview_did="did:web:api.bsky.app",
-		)
-		db.create_account(
-			did=TEST_DID,
-			handle=TEST_HANDLE,
-			password=TEST_PASSWORD,
-			privkey=TEST_PRIVKEY,
-		)
+			service_run_task = asyncio.create_task(
+				service_run_and_capture_port(
+					queue,
+					db=db,
+					client=client,
+					sock_path=None,
+					host="localhost",
+					port=0,
+				)
+			)
+			queue_get_task = asyncio.create_task(queue.get())
+			done, pending = await asyncio.wait(
+				(queue_get_task, service_run_task),
+				return_when=asyncio.FIRST_COMPLETED,
+			)
+			if done == service_run_task:
+				raise service_run_task.execption()
+			else:
+				port = queue_get_task.result()
 
-		try:
-			yield TestPDS(
-				endpoint=f"http://{hostname}",
-				db=db,
+			hostname = f"localhost:{port}"
+			db.update_config(
+				pds_pfx=f'http://{hostname}',
+				pds_did=f'did:web:{urllib.parse.quote(hostname)}',
+				bsky_appview_pfx="https://api.bsky.app",
+				bsky_appview_did="did:web:api.bsky.app",
 			)
-		finally:
-			service_run_task.cancel()
+			db.create_account(
+				did=TEST_DID,
+				handle=TEST_HANDLE,
+				password=TEST_PASSWORD,
+				privkey=TEST_PRIVKEY,
+			)
+
 			try:
-				await service_run_task
-			except asyncio.CancelledError:
-				pass
+				yield PDSInfo(
+					endpoint=f"http://{hostname}",
+					db=db,
+				)
+			finally:
+				db.con.close()
+				service_run_task.cancel()
+				try:
+					await service_run_task
+				except asyncio.CancelledError:
+					pass
 
 @pytest.fixture
 async def s(aiolib):