Source code for uranium.rules.was_changed

import os
import time
from .base import RuleBase
KEY = "_uranium.rules.was_changed"

[docs]class WasChanged(RuleBase): """ WasChanged is a rule that activates if the task has never run, or if a path has a file modified since the task last ran. .. code:: python import subprocess from uranium import rule from uranium.rules import WasChanged # only run tests if the code changed. @rule(WasChanged("./my_module")) def test(build): build.packages.install("pytest") return["py.test", build.root]) """ def __init__(self, path): self._path = path def path(self, build): return os.path.join(build.root, self._path) def before(self, build): previous_timestamp = build.history.get(self.key) if not previous_timestamp: return False current_timestamp = self._get_timestamp(self.path(build)) return current_timestamp <= previous_timestamp def after(self, build): current_timestamp = self._get_timestamp(self.path(build)) build.history[self.key] = current_timestamp @property def key(self): return "{0}.{1}.{2}".format( KEY, self._path, self.func.__name__ ) @classmethod def _get_timestamp(cls, path): """ return the timestamp as a UTC datetime object. """ if os.path.isdir(path): return cls._get_dir_timestamp(path) else: return os.path.getmtime(path) @classmethod def _get_dir_timestamp(cls, path): oldest_timestamp = time.time() for root, _, filename in os.walk(path): f_path = os.path.join(root, filename) ts = os.path.getmtime(f_path) if ts < oldest_timestamp: oldest_timestamp = ts return oldest_timestamp