Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/debsbom/commands/download.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
from zstandard import ZstdCompressor, ZstdDecompressor
import requests
from ..snapshot import client as sdlclient
from ..download.adapters import LocalFileAdapter
from ..download.download import PackageDownloader
from ..download.resolver import PersistentResolverCache, UpstreamResolver
from debsbom.download.download import DownloadStatus, DownloadResult
Expand Down Expand Up @@ -77,6 +78,7 @@ def run(cls, args):
else:
resolver = cls.get_pkgstream_resolver()
rs = requests.Session()
rs.mount("file:///", LocalFileAdapter())
rs.headers.update({"User-Agent": f"debsbom/{version('debsbom')}"})
sdl = sdlclient.SnapshotDataLake(session=rs)
u_resolver = UpstreamResolver(sdl, cache)
Expand Down
44 changes: 44 additions & 0 deletions src/debsbom/download/adapters.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
# Copyright (C) 2025 Siemens
#
# SPDX-License-Identifier: MIT

import errno
from io import BytesIO
import locale
from pathlib import Path
from requests import Response, Request, Session, codes
from requests.adapters import BaseAdapter
from urllib.parse import unquote, urlparse


class LocalFileAdapter(BaseAdapter):
"""Adapter for local file access."""

def send(self, request, **kwargs) -> Response:
if request.method != "GET":
raise ValueError(f"Request method {request.method} is not supported")

response = Response()
response.request = request
response.url = request.url

path = Path(unquote(urlparse(request.url).path))
try:
response.raw = open(path, "rb")
# make sure we properly close the file when we are done
response.raw.release_conn = response.raw.close
response.status_code = codes.ok
except IOError as e:
if e.errno == errno.EACCES:
response.status_code = codes.forbidden
elif e.errno == errno.ENOENT:
response.status_code = codes.not_found
else:
response.status_code = codes.bad_request
response.raw = BytesIO(str(e).encode(locale.getpreferredencoding()))
response.reason = str(e)

return response

def close(self):
pass
1 change: 1 addition & 0 deletions tests/data/local-download
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
This is a test file for the local file adapter test.
17 changes: 17 additions & 0 deletions tests/test_download.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import jsonschema

import pytest
from debsbom.download.adapters import LocalFileAdapter
from debsbom.download import (
PackageDownloader,
PersistentResolverCache,
Expand All @@ -31,6 +32,7 @@
import spdx_tools.spdx.writer.json.json_writer as spdx_json_writer
import cyclonedx.output as cdx_output
import cyclonedx.schema as cdx_schema
from requests import Session

from unittest import mock

Expand Down Expand Up @@ -291,3 +293,18 @@ def test_download_result_invalid(dlschema):
}
with pytest.raises(jsonschema.ValidationError):
jsonschema.validate(data, schema=dlschema)


def test_local_file():
session = Session()
session.mount("file:///", LocalFileAdapter())
with session.get("file://" + str(Path("tests/data/local-download").absolute())) as r:
assert r.status_code == 200
assert r.content == b"This is a test file for the local file adapter test.\n"


def test_local_file_404():
session = Session()
session.mount("file:///", LocalFileAdapter())
with session.get("file:///does-not-exist") as r:
assert r.status_code == 404