summaryrefslogtreecommitdiff
path: root/tests/frontend/project/sources/fetch_source.py
blob: c62d1d29a202dfa23a628597f82c9111c41d267b (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import os

from buildstream import Source, SourceError, SourceFetcher

# Expected config
# sources:
# - output-text: $FILE
#   urls:
#   - foo:bar
#   - baz:quux
#   fetch-succeeds:
#     Foo/bar: true
#     ooF/bar: false


class FetchFetcher(SourceFetcher):
    def __init__(self, source, url, primary=False):
        super().__init__()
        self.source = source
        self.original_url = url
        self.primary = primary
        self.mark_download_url(url)

    def fetch(self, alias_override=None):
        url = self.source.translate_url(self.original_url, alias_override=alias_override, primary=self.primary)
        with open(self.source.output_file, "a") as f:
            success = url in self.source.fetch_succeeds and self.source.fetch_succeeds[url]
            message = "Fetch {} {} from {}\n".format(self.original_url, "succeeded" if success else "failed", url)
            f.write(message)
            if not success:
                raise SourceError("Failed to fetch {}".format(url))


class FetchSource(Source):
    # Read config to know which URLs to fetch
    def configure(self, node):
        self.original_urls = node.get_str_list("urls")
        self.output_file = node.get_str("output-text")
        self.fetch_succeeds = {key: value.as_bool() for key, value in node.get_mapping("fetch-succeeds", {}).items()}

        # First URL is the primary one for this test
        #
        primary = True
        self.fetchers = []
        for url in self.original_urls:
            self.mark_download_url(url, primary=primary)
            fetcher = FetchFetcher(self, url, primary=primary)
            self.fetchers.append(fetcher)
            primary = False

    def get_source_fetchers(self):
        return self.fetchers

    def preflight(self):
        output_dir = os.path.dirname(self.output_file)
        if not os.path.exists(output_dir):
            raise SourceError("Directory '{}' does not exist".format(output_dir))

    def stage(self, directory):
        pass

    def fetch(self):
        for fetcher in self.fetchers:
            fetcher.fetch()

    def get_unique_key(self):
        return {"urls": self.original_urls, "output_file": self.output_file}

    def is_resolved(self):
        return True

    def is_cached(self) -> bool:
        if not os.path.exists(self.output_file):
            return False

        with open(self.output_file, "r") as f:
            contents = f.read()
            for url in self.original_urls:
                if url not in contents:
                    return False

        return True

    # We dont have a ref, we're a local file...
    def load_ref(self, node):
        pass

    def get_ref(self):
        return None  # pragma: nocover

    def set_ref(self, ref, node):
        pass  # pragma: nocover


def setup():
    return FetchSource