-
Notifications
You must be signed in to change notification settings - Fork 27
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Signed-off-by: Konrad Weihmann <[email protected]>
- Loading branch information
1 parent
eb259df
commit 6ab636c
Showing
6 changed files
with
225 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,71 @@ | ||
import argparse | ||
import hashlib | ||
import os | ||
import pickle # noqa: S403 | ||
import shutil | ||
from typing import Any, List, Tuple, Union | ||
|
||
__default_cache_dir = os.path.join(os.environ.get('HOME'), '.oelint/caches') | ||
|
||
|
||
class Caches(): | ||
|
||
def __init__(self, args: argparse.Namespace) -> None: | ||
self.__directory = args.cachedir | ||
self.__enabled = args.cached | ||
self.__quiet = args.quiet | ||
self.__arg_fingerprint = self.__calculate_args_fingerprint(args) | ||
|
||
if self.__enabled: | ||
os.makedirs(self.__directory, exist_ok=True) | ||
|
||
@property | ||
def FingerPrint(self): | ||
return self.__arg_fingerprint | ||
|
||
def __calculate_args_fingerprint(self, args: argparse.Namespace) -> object: | ||
_hash = hashlib.sha1(usedforsecurity=False) | ||
for item in [ | ||
# list all hashing relevant arguments | ||
args.suppress, | ||
args.relpaths, | ||
args.state.hide, | ||
args.release, | ||
args.color, | ||
]: | ||
_hash.update(f'{item}'.encode()) | ||
return _hash.hexdigest() | ||
|
||
def AddToFingerPrint(self, input_: Union[str, bytes]) -> None: | ||
_hash = hashlib.sha1(self.__arg_fingerprint.encode(), usedforsecurity=False) # noqa: DUO130 | ||
if isinstance(input_, str): | ||
input_ = input_.encode() | ||
_hash.update(input_) | ||
self.__arg_fingerprint = _hash.hexdigest() | ||
|
||
def ClearCaches(self) -> None: | ||
shutil.rmtree(self.__directory, ignore_errors=True) | ||
|
||
def GetFromCache(self, rule_ids: List[str], stash_fingerprint: str) -> Union[None, Tuple[Tuple[str, int], List[str], str]]: | ||
if not self.__enabled: | ||
return None | ||
_hash = hashlib.sha1(f'{self.__arg_fingerprint}{rule_ids}{stash_fingerprint}'.encode(), usedforsecurity=False) # noqa: DUO130 | ||
_hash_path = os.path.join(self.__directory, _hash.hexdigest()) | ||
try: | ||
with open(_hash_path, 'rb') as i: # pragma: no cover | ||
if not self.__quiet: | ||
print(f'Using cached item {_hash_path}') | ||
return pickle.load(i) # noqa: DUO103, S301 | ||
except (pickle.PickleError, FileNotFoundError): | ||
return None | ||
|
||
def SaveToCache(self, rule_ids: List[str], stash_fingerprint: str, content: Any) -> None: | ||
if not self.__enabled: | ||
return | ||
_hash = hashlib.sha1(f'{self.__arg_fingerprint}{rule_ids}{stash_fingerprint}'.encode(), usedforsecurity=False) # noqa: DUO130 | ||
_hash_path = os.path.join(self.__directory, _hash.hexdigest()) | ||
try: | ||
with open(_hash_path, 'wb') as o: | ||
return pickle.dump(content, o, protocol=0) | ||
except (pickle.PicklingError, FileNotFoundError): # pragma: no cover | ||
pass # pragma: no cover |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,103 @@ | ||
import pytest # noqa: I900 | ||
import os | ||
import glob | ||
|
||
from .base import TestBaseClass | ||
|
||
|
||
# flake8: noqa S101 - n.a. for test files | ||
class TestCached(TestBaseClass): | ||
|
||
@pytest.mark.parametrize('input_', | ||
[ | ||
{ | ||
'test.bb': | ||
''' | ||
INSANE_SKIP:${PN} = "foo" | ||
''', | ||
}, | ||
], | ||
) | ||
def test_cached(self, capsys, input_): | ||
|
||
from oelint_adv.core import run | ||
from oelint_adv.__main__ import arguments_post | ||
|
||
tmpdir = os.path.dirname(self._create_tempfile('tmppath/.marker', '')) | ||
files = [self._create_tempfile(k, v) for k, v in input_.items()] | ||
|
||
args = arguments_post(self._create_args_parser().parse_args( | ||
['--jobs=1', '--cached', f'--cachedir={tmpdir}', *files] | ||
)) | ||
|
||
run(args) | ||
|
||
captured = capsys.readouterr() | ||
|
||
assert 'Using cached item ' not in captured.out | ||
|
||
assert any(glob.glob(f'{tmpdir}/*')) | ||
|
||
run(args) | ||
|
||
@pytest.mark.parametrize('input_', | ||
[ | ||
{ | ||
'test.bb': | ||
''' | ||
INSANE_SKIP:${PN} = "foo" | ||
''', | ||
}, | ||
], | ||
) | ||
def test_no_cached(self, capsys, input_): | ||
|
||
from oelint_adv.core import run | ||
from oelint_adv.__main__ import arguments_post | ||
|
||
tmpdir = os.path.dirname(self._create_tempfile('tmppath/.marker', '')) | ||
files = [self._create_tempfile(k, v) for k, v in input_.items()] | ||
|
||
args = arguments_post(self._create_args_parser().parse_args( | ||
['--jobs=1', *files] | ||
)) | ||
|
||
run(args) | ||
|
||
captured = capsys.readouterr() | ||
|
||
assert 'Using cached item ' not in captured.out | ||
|
||
assert not any(glob.glob(f'{tmpdir}/*')) | ||
|
||
def test_clear_cached(self): | ||
|
||
from oelint_adv.__main__ import arguments_post | ||
|
||
tmpdir = os.path.dirname(self._create_tempfile('tmppath/.marker', '')) | ||
|
||
args = arguments_post(self._create_args_parser().parse_args( | ||
['--jobs=1', '--cached', f'--cachedir={tmpdir}', '--clear-caches'] | ||
)) | ||
|
||
args.state._caches.ClearCaches() | ||
|
||
def test_arg_fingerprint(self): | ||
|
||
from oelint_adv.__main__ import arguments_post | ||
|
||
tmpdir = os.path.dirname(self._create_tempfile('tmppath/.marker', '')) | ||
|
||
args = arguments_post(self._create_args_parser().parse_args( | ||
['--jobs=1', '--cached', f'--cachedir={tmpdir}', self._create_tempfile('test.bb', '')] | ||
)) | ||
|
||
fp = args.state._caches.FingerPrint | ||
|
||
args.state._caches.AddToFingerPrint(b'1234') | ||
|
||
assert args.state._caches.FingerPrint != fp | ||
|
||
args.state._caches.AddToFingerPrint('1234') | ||
|
||
assert args.state._caches.FingerPrint != fp |