summaryrefslogtreecommitdiffstats
path: root/INFO.yaml
blob: cb4499644c26001d3a84eb7075325f939bb18229 (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

# ============LICENSE_START=======================================================
# Copyright (C) 2021-2022 Nordix Foundation.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# SPDX-License-Identifier: Apache-2.0
# ============LICENSE_END=========================================================

---
project: 'cps'
project_creation_date: '2020-10-27'
lifecycle_state: 'Incubation'
project_category: ''
project_lead: &onap_releng_ptl
    name: 'Toine Siebelink'
    email: 'toine.siebelink@est.tech'
    id: 'ToineSiebelink'
    company: 'Ericsson Software Technology'
    timezone: 'Europe/Dublin'
primary_contact: *onap_releng_ptl
issue_tracking:
    type: 'jira'
    url: 'https://jira.onap.org/projects/CPS'
    key: 'CPS'
mailing_list:
    type: 'groups.io'
    url: 'lists.onap.org'
    tag: '<[sub-project_name]>'
realtime_discussion: ''
meetings:
    - type: 'zoom'
      agenda: 'https://wiki.onap.org/pages/viewpage.action?pageId=111117075'
      url: 'https://zoom.us/j/836561560?pwd=TTZNcFhXTWYxMmZ4SlgzcVZZQXluUT09'
      server: 'n/a'
      channel: 'n/a'
      repeats: 'weekly'
      time: '08:00 America/New_York'
committers:
    - <<: *onap_releng_ptl
    - name: 'Sourabh Sharma'
      email: 'sourabh.sourabh@est.tech'
      company: 'Ericsson Software Technology'
      id: 'sourabh_sourabh'
      timezone: 'Europe/Dublin'
    - name: 'Luke Gleeson'
      email: 'luke.gleeson@est.tech'
      company: 'Ericsson Software Technology'
      id: 'lukegleeson'
      timezone: 'Europe/Dublin'
    - name: 'Priyank Maheshwari'
      email: 'priyank.maheshwari@est.tech'
      company: 'Ericsson Software Technology'
      id: 'mpriyank'
      timezone: 'Europe/Dublin'
repositories:
    - cps
tsc:
    approval: 'https://lists.onap.org/g/onap-tsc'
">name: str, version: str) -> bytes: """Download blueprint file from repository. :param name: Blueprint name :param version: Blueprint version :return: Zipped Blueprint file bytes """ @abstractmethod def remove_blueprint(self, name: str, version: str) -> None: """Remove blueprint file from repository. :param name: Blueprint name :param version: Blueprint version """ class FileRepository(Repository): """Store blueprints on local directory.""" base_path = None def __init__(self, base_path: Path) -> None: """Initialize the repository while passing the needed path. :param base_path: Local OS path on which blueprint files reside. """ self.base_path = base_path def __remove_directory_tree(self, full_path: str) -> None: """Remove specified path. :param full_path: Full path to a directory. :raises: FileNotFoundError """ try: shutil.rmtree(full_path, ignore_errors=False) except OSError: raise ArtifactNotFoundError def __create_directory_tree(self, full_path: str, mode: int = 0o744, retry_on_error: bool = True) -> None: """Create directory or overwrite existing one. This method will handle a directory tree creation. If there is a collision in directory structure - old directory tree will be removed and creation will be attempted one more time. If the creation fails for the second time the exception will be raised. :param full_path: Full directory tree path (eg. one/two/tree) as string. :param mode: Permission mask for the directories. :param retry_on_error: Flag that indicates if there should be a attempt to retry the operation. """ try: os.makedirs(full_path, mode=mode) except FileExistsError: # In this case we know that cba of same name and version need to be overwritten if retry_on_error: self.__remove_directory_tree(full_path) self.__create_directory_tree(full_path, mode=mode, retry_on_error=False) else: # This way we won't try for ever if something goes wrong raise ArtifactOverwriteError def upload_blueprint(self, cba_bytes: bytes, name: str, version: str) -> None: """Store blueprint file in the repository. :param cba_bytes: Bytes to save :param name: Blueprint name :param version: Blueprint version """ temporary_file: BytesIO = BytesIO(cba_bytes) if not is_zipfile(temporary_file): raise InvalidRequestError target_path: str = str(Path(self.base_path.absolute(), name, version)) self.__create_directory_tree(target_path) with ZipFile(temporary_file, "r") as zip_file: # type: ZipFile zip_file.extractall(target_path) def download_blueprint(self, name: str, version: str) -> bytes: """Download blueprint file from repository. This method does the in-memory zipping the files and returns bytes :param name: Blueprint name :param version: Blueprint version :return: Zipped Blueprint file bytes """ temporary_file: BytesIO = BytesIO() files_path: str = str(Path(self.base_path.absolute(), name, version)) if not os.path.exists(files_path): raise ArtifactNotFoundError with ZipFile(temporary_file, "w") as zip_file: # type: ZipFile for directory_name, subdirectory_names, filenames in os.walk(files_path): # type: str, list, list for filename in filenames: # type: str zip_file.write(Path(directory_name, filename)) # Rewind the fake file to allow reading temporary_file.seek(0) zip_as_bytes: bytes = temporary_file.read() temporary_file.close() return zip_as_bytes def remove_blueprint(self, name: str, version: str) -> None: """Remove blueprint file from repository. :param name: Blueprint name :param version: Blueprint version :raises: FileNotFoundError """ files_path: str = str(Path(self.base_path.absolute(), name, version)) self.__remove_directory_tree(files_path) class RepositoryStrategy(ABC): """Strategy class. It has only one public method `get_repository`, which returns valid repository instance for the the configuration value. You can create many Repository subclasses, but repository clients doesn't have to know which one you use. """ @classmethod def get_reporitory(cls) -> Repository: """Get the valid repository instance for the configuration value. Currently it returns FileRepository because it is an only Repository implementation. """ return FileRepository(Path(config["artifactManagerServer"]["fileRepositoryBasePath"]))