492 lines
18 KiB
Python
492 lines
18 KiB
Python
#!/usr/bin/python
|
|
|
|
# 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.
|
|
|
|
import argparse
|
|
import logging
|
|
import pathlib
|
|
import requests
|
|
import subprocess
|
|
|
|
from git import exc
|
|
from git import Repo
|
|
|
|
from jinja2 import PackageLoader
|
|
from jinja2 import Environment
|
|
from jinja2 import select_autoescape
|
|
|
|
import otc_metadata.services
|
|
|
|
data = otc_metadata.services.Services()
|
|
|
|
api_session = requests.Session()
|
|
|
|
|
|
def process_repositories(args, service):
|
|
"""Checkout repositories"""
|
|
logging.debug(f"Processing service {service}")
|
|
workdir = pathlib.Path(args.work_dir)
|
|
workdir.mkdir(exist_ok=True)
|
|
|
|
copy_to = None
|
|
repo_to = None
|
|
url_to = None
|
|
target_repo = None
|
|
git_fqdn = None
|
|
error_list = []
|
|
|
|
env = Environment(
|
|
loader=PackageLoader("otc_metadata"), autoescape=select_autoescape()
|
|
)
|
|
conf_py_template = env.get_template("conf.py.j2")
|
|
tox_ini_template = env.get_template("tox.ini.j2")
|
|
zuul_yaml_template = env.get_template("zuul.yaml.j2")
|
|
index_sbv_template = env.get_template("index_sbv.rst.j2")
|
|
doc_requirements_template = env.get_template("doc_requirements.txt.j2")
|
|
|
|
for repo in service["repositories"]:
|
|
if repo["cloud_environments"][0] == args.cloud_environment:
|
|
logging.debug(f"Processing repository {repo}")
|
|
repo_dir = pathlib.Path(workdir, repo["type"], repo["repo"])
|
|
|
|
if repo["environment"] == args.target_environment:
|
|
copy_to = repo_dir
|
|
else:
|
|
logging.debug(f"Skipping repository {repo}")
|
|
continue
|
|
|
|
repo_dir.mkdir(parents=True, exist_ok=True)
|
|
if repo["type"] == "gitea":
|
|
repo_url = (
|
|
f"ssh://git@gitea.eco.tsi-dev.otc-service.com:2222/"
|
|
f"{repo['repo']}"
|
|
)
|
|
git_fqdn = "gitea.eco.tsi-dev.otc-service.com"
|
|
elif repo["type"] == "github":
|
|
repo_url = f"git@github.com:/{repo['repo']}"
|
|
else:
|
|
logging.error(f"Repository type {repo['type']} is not supported")
|
|
error_list.append({
|
|
"error": f"Repository type {repo['type']} is not supported",
|
|
"repo": repo['repo']
|
|
})
|
|
continue
|
|
|
|
if repo_dir.exists():
|
|
logging.debug(f"Repository {repo} already checked out")
|
|
try:
|
|
git_repo = Repo(repo_dir)
|
|
git_repo.remotes.origin.fetch()
|
|
git_repo.heads.main.checkout()
|
|
git_repo.remotes.origin.pull()
|
|
except exc.InvalidGitRepositoryError:
|
|
logging.error("Existing repository checkout is bad")
|
|
repo_dir.rmdir()
|
|
except Exception as e:
|
|
error_list.append({
|
|
"error": e,
|
|
"repo": repo['repo']
|
|
})
|
|
|
|
if not repo_dir.exists():
|
|
try:
|
|
git_repo = Repo.clone_from(repo_url, repo_dir, branch="main")
|
|
except Exception:
|
|
logging.error(f"Error cloning repository {repo_url}")
|
|
error_list.append({
|
|
"error": f"Error cloning repository {repo_url}",
|
|
"repo": repo['repo']
|
|
})
|
|
continue
|
|
|
|
if repo["environment"] == args.target_environment:
|
|
url_to = repo_url
|
|
repo_to = git_repo
|
|
target_repo = repo
|
|
break
|
|
|
|
if not target_repo:
|
|
logging.info(
|
|
f"No repository service {service['service_title']}"
|
|
f"for environment {args.target_environment} in cloud_environment {args.cloud_environment}"
|
|
)
|
|
return
|
|
|
|
branch_name = f"{args.branch_name}"
|
|
if args.branch_force:
|
|
logging.debug("Dropping current branch")
|
|
try:
|
|
repo_to.delete_head(branch_name, force=True)
|
|
repo_to.delete_head(f"refs/heads/{args.branch_name}", force=True)
|
|
except exc.GitCommandError as e:
|
|
print(e)
|
|
error_list.append({
|
|
"error": e,
|
|
"repo": target_repo['repo']
|
|
})
|
|
pass
|
|
try:
|
|
new_branch = repo_to.create_head(branch_name, "main")
|
|
except Exception as e:
|
|
logging.warning(f"Skipping service {service} due to {e}")
|
|
error_list.append({
|
|
"error": e,
|
|
"repo": target_repo['repo']
|
|
})
|
|
return
|
|
new_branch.checkout()
|
|
|
|
service_docs = list(data.docs_by_service_type(service["service_type"]))
|
|
|
|
for doc in service_docs:
|
|
for cloud_environment in doc["cloud_environments"]:
|
|
if cloud_environment["name"] == args.cloud_environment:
|
|
logging.debug(f"Analyzing document {doc}")
|
|
|
|
conf_py_path = pathlib.Path(copy_to, doc["rst_location"], "conf.py")
|
|
if not conf_py_path.exists():
|
|
logging.info(f"Path for document {doc['title']} does not exist")
|
|
conf_py_path.parent.mkdir(parents=True, exist_ok=True)
|
|
context = dict(
|
|
repo_name=target_repo["repo"],
|
|
project=service["service_title"],
|
|
title=f"{service['service_title']} - {doc['title']}",
|
|
)
|
|
if "pdf_name" in doc:
|
|
context["pdf_name"] = doc["pdf_name"]
|
|
if git_fqdn:
|
|
context["git_fqdn"] = git_fqdn
|
|
if target_repo.get("type") != "github":
|
|
context["git_type"] = target_repo["type"]
|
|
if (args.target_environment == "public"
|
|
and args.cloud_environment == "swiss"):
|
|
context["html_options"] = dict(
|
|
logo_url="https://docs.sc.otc.t-systems.com",
|
|
)
|
|
elif (args.target_environment == "public"):
|
|
context["html_options"] = dict(
|
|
logo_url="https://docs.otc.t-systems.com",
|
|
)
|
|
elif (args.target_environment == "internal"
|
|
and args.cloud_environment == "swiss"):
|
|
context["html_options"] = dict(
|
|
disable_search=True,
|
|
site_name="Internal Documentation Portal",
|
|
logo_url="https://docs-swiss-int.otc-service.com",
|
|
)
|
|
elif args.target_environment == "internal":
|
|
context["html_options"] = dict(
|
|
disable_search=True,
|
|
site_name="Internal Documentation Portal",
|
|
logo_url="https://docs-int.otc-service.com",
|
|
)
|
|
context["doc_environment"] = args.target_environment
|
|
if doc['link']:
|
|
context["doc_link"] = doc['link']
|
|
else:
|
|
context["doc_link"] = (
|
|
'/'
|
|
+ service['service_uri']
|
|
+ '/'
|
|
+ doc['type']
|
|
+ '/'
|
|
)
|
|
context["doc_title"] = doc['title']
|
|
context["doc_type"] = doc['type']
|
|
context["service_category"] = service['service_category']
|
|
context["service_title"] = service['service_title']
|
|
context["service_type"] = service['service_type']
|
|
|
|
conf_py_content = conf_py_template.render(**context)
|
|
with open(conf_py_path, "w", encoding="utf-8", newline="") as out:
|
|
logging.debug(f"Generating {conf_py_path} from template...")
|
|
out.write(conf_py_content)
|
|
|
|
repo_to.index.add([doc["rst_location"]])
|
|
|
|
if args.update_sbv:
|
|
"""Add or update service-based-view"""
|
|
copy_path = pathlib.Path(copy_to, 'doc', 'source')
|
|
context = dict(
|
|
repo_name=target_repo["repo"],
|
|
project=service["service_title"],
|
|
# pdf_name=doc["pdf_name"],
|
|
title=f"{service['service_title']} - Service Based View",
|
|
service_type=service["service_type"]
|
|
)
|
|
context["service_category"] = service['service_category']
|
|
context["service_title"] = service['service_title']
|
|
if not copy_path.exists():
|
|
logging.info("Path for sbv does not exist")
|
|
copy_path.mkdir(parents=True, exist_ok=True)
|
|
context["otc_sbv"] = True
|
|
if git_fqdn:
|
|
context["git_fqdn"] = git_fqdn
|
|
if target_repo.get("type") != "github":
|
|
context["git_type"] = target_repo["type"]
|
|
if (args.target_environment == "public"
|
|
and args.cloud_environment == "swiss"):
|
|
context["html_options"] = dict(
|
|
logo_url="https://docs.sc.otc.t-systems.com",
|
|
)
|
|
elif (args.target_environment == "public"):
|
|
context["html_options"] = dict(
|
|
logo_url="https://docs.otc.t-systems.com",
|
|
)
|
|
elif (args.target_environment == "internal"
|
|
and args.cloud_environment == "swiss"):
|
|
context["html_options"] = dict(
|
|
disable_search=True,
|
|
site_name="Internal Documentation Portal",
|
|
logo_url="https://docs-swiss-int.otc-service.com",
|
|
)
|
|
elif args.target_environment == "internal":
|
|
context["html_options"] = dict(
|
|
disable_search=True,
|
|
site_name="Internal Documentation Portal",
|
|
logo_url="https://docs-int.otc-service.com",
|
|
)
|
|
context["environment"] = args.target_environment
|
|
sbv_title = (service["service_title"] + "\n"
|
|
+ ('=' * len(service["service_title"])))
|
|
context["sbv_title"] = sbv_title
|
|
conf_py_content = conf_py_template.render(**context)
|
|
index_sbv_content = index_sbv_template.render(**context)
|
|
with open(
|
|
pathlib.Path(copy_path, "conf.py"),
|
|
"w",
|
|
encoding="utf-8") as out:
|
|
out.write(conf_py_content)
|
|
repo_to.index.add(pathlib.Path(copy_path, "conf.py"))
|
|
|
|
if (not args.overwrite_index_sbv
|
|
and pathlib.Path(copy_path, "index.rst").exists()):
|
|
logging.info("File index.rst for sbv exists. Skipping")
|
|
else:
|
|
with open(
|
|
pathlib.Path(copy_path, "index.rst"),
|
|
"w",
|
|
encoding="utf-8") as out:
|
|
out.write(index_sbv_content)
|
|
repo_to.index.add(pathlib.Path(copy_path, "index.rst"))
|
|
|
|
placeholder_path = pathlib.Path(copy_path, "_static")
|
|
if not pathlib.Path(placeholder_path, "placeholder").exists():
|
|
placeholder_path.mkdir(parents=True, exist_ok=True)
|
|
open(pathlib.Path(placeholder_path, "placeholder"), 'a').close()
|
|
repo_to.index.add(pathlib.Path(placeholder_path, "placeholder"))
|
|
|
|
if args.update_tox:
|
|
"""Update tox.ini"""
|
|
context = dict(docs=[])
|
|
for doc in service_docs:
|
|
for cloud_environment in doc["cloud_environments"]:
|
|
if cloud_environment["name"] == args.cloud_environment:
|
|
if doc["type"] == "dev":
|
|
doc["type"] = "dev-guide"
|
|
context["docs"].append(doc)
|
|
context["pdf_enabled"] = cloud_environment["pdf_enabled"]
|
|
|
|
context["target_environment"] = args.target_environment
|
|
context["service_type"] = service['service_type']
|
|
|
|
tox_ini_content = tox_ini_template.render(**context)
|
|
tox_ini_path = pathlib.Path(copy_to, "tox.ini")
|
|
doc_requirements_content = doc_requirements_template.render(**context)
|
|
doc_requirements_path = pathlib.Path(
|
|
copy_to, "doc", "requirements.txt"
|
|
)
|
|
doc_requirements_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(tox_ini_path, "w", encoding="utf-8", newline="") as out:
|
|
logging.debug(f"Generating {tox_ini_path} from template...")
|
|
out.write(tox_ini_content)
|
|
repo_to.index.add(["tox.ini"])
|
|
with open(
|
|
doc_requirements_path, "w", encoding="utf-8", newline=""
|
|
) as out:
|
|
logging.debug(
|
|
f"Generating {doc_requirements_path} from template..."
|
|
)
|
|
out.write(doc_requirements_content)
|
|
repo_to.index.add(["doc/requirements.txt"])
|
|
|
|
if args.update_zuul:
|
|
"""Update zuul.yaml"""
|
|
context = dict(docs=[])
|
|
for doc in service_docs:
|
|
for cloud_environment in doc["cloud_environments"]:
|
|
if cloud_environment["name"] == args.cloud_environment:
|
|
if doc["type"] == "dev":
|
|
doc["type"] = "dev-guide"
|
|
context["docs"].append(doc)
|
|
context["service_type"] = service['service_type']
|
|
|
|
zuul_yaml_content = zuul_yaml_template.render(**context)
|
|
zuul_yaml_path = pathlib.Path(copy_to, "zuul.yaml")
|
|
with open(zuul_yaml_path, "w", encoding="utf-8", newline="") as out:
|
|
logging.debug(f"Generating {zuul_yaml_path} from template...")
|
|
out.write(zuul_yaml_content)
|
|
repo_to.index.add(["zuul.yaml"])
|
|
|
|
if len(repo_to.index.diff("HEAD")) == 0:
|
|
# Nothing to commit
|
|
logging.debug(
|
|
"No changes for service %s required" % service["service_type"]
|
|
)
|
|
return
|
|
repo_to.index.commit(
|
|
args.commit_description
|
|
)
|
|
push_args = ["--set-upstream", "origin", branch_name]
|
|
if args.force_push:
|
|
push_args.append("--force")
|
|
try:
|
|
repo_to.git.push(*push_args)
|
|
except Exception as e:
|
|
error_list.append({
|
|
"error": e,
|
|
"repo": repo['repo']
|
|
})
|
|
if "github" in url_to:
|
|
subprocess.run(
|
|
args=["gh", "pr", "create", "-f"], cwd=copy_to, check=False
|
|
)
|
|
elif "gitea" in url_to and args.token:
|
|
open_pr(
|
|
args,
|
|
repo["repo"],
|
|
dict(
|
|
title="Update Docs configuration",
|
|
head=branch_name,
|
|
),
|
|
)
|
|
if len(error_list) != 0:
|
|
logging.error("The following errors have happened:")
|
|
logging.error(error_list)
|
|
|
|
|
|
def open_pr(args, repository, pr_data):
|
|
req = dict(
|
|
base=pr_data.get("base", "main"),
|
|
head=pr_data["head"],
|
|
)
|
|
if "title" in pr_data:
|
|
req["title"] = pr_data["title"]
|
|
if "body" in pr_data:
|
|
req["body"] = pr_data["body"].replace("\\n", "\n")
|
|
if "assignees" in pr_data:
|
|
req["assignees"] = pr_data["assignees"]
|
|
if "labels" in pr_data:
|
|
req["labels"] = pr_data["labels"]
|
|
rsp = api_session.post(
|
|
f"{args.api_url}/repos/{repository}/pulls", json=req
|
|
)
|
|
if rsp.status_code != 201:
|
|
print(rsp.text)
|
|
# print(f"Going to open PR with title {pr_data['title']} in {repository}")
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Update conf.py file in repositories."
|
|
)
|
|
parser.add_argument(
|
|
"--target-environment",
|
|
required=True,
|
|
choices=["internal", "public"],
|
|
help="Environment to be used as a source",
|
|
)
|
|
parser.add_argument("--service-type", help="Service to update")
|
|
parser.add_argument(
|
|
"--update-tox", action="store_true", help="Whether to update tox.ini."
|
|
)
|
|
parser.add_argument(
|
|
"--work-dir",
|
|
required=True,
|
|
help="Working directory to use for repository checkout.",
|
|
)
|
|
parser.add_argument(
|
|
"--branch-name",
|
|
default="confpy",
|
|
help="Branch name to be used for synchronizing.",
|
|
)
|
|
parser.add_argument(
|
|
"--branch-force",
|
|
action="store_true",
|
|
help="Whether to force branch recreation.",
|
|
)
|
|
parser.add_argument("--token", metavar="token", help="API token")
|
|
parser.add_argument("--api-url", help="API base url of the Git hoster")
|
|
parser.add_argument(
|
|
"--update-sbv",
|
|
action="store_true",
|
|
help="Whether to update service-based-view"
|
|
)
|
|
parser.add_argument(
|
|
"--update-zuul",
|
|
action="store_true",
|
|
help="Whether to update zuul.yaml"
|
|
)
|
|
parser.add_argument(
|
|
"--overwrite-index-sbv",
|
|
action="store_true",
|
|
help=("Whether to overwrite index.rst for service-based-view."
|
|
+ "\nCan only be used if --update-sbv is also specified")
|
|
)
|
|
parser.add_argument(
|
|
"--force-push",
|
|
action="store_true",
|
|
help="Whether to force push the commit"
|
|
)
|
|
parser.add_argument(
|
|
"--commit-description",
|
|
default=(
|
|
"Update tox.ini && conf.py file\n\n"
|
|
"Performed-by: gitea/infra/otc-metadata/"
|
|
"tools/generate_doc_confpy.py"
|
|
),
|
|
help="Commit description for the commit",
|
|
)
|
|
parser.add_argument(
|
|
"--cloud-environment",
|
|
required=True,
|
|
default="eu_de",
|
|
help="Cloud Environment. Default: eu_de",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
services = []
|
|
if args.overwrite_index_sbv and not args.update_sbv:
|
|
logging.error(
|
|
"Cannot overwrite index.rst for service-based-view"
|
|
+ " without updating service-based-view"
|
|
)
|
|
exit(1)
|
|
|
|
if args.service_type:
|
|
services = [data.get_service_with_repo_by_service_type(service_type=args.service_type)]
|
|
else:
|
|
services = data.services_with_repos()
|
|
|
|
if args.token:
|
|
api_session.headers.update({"Authorization": f"token {args.token}"})
|
|
|
|
for service in services:
|
|
process_repositories(args, service)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|