feat(qa_check): enable checking connector docs structure via qa check

airbyte-ci/connectors/connector_ops/connector_ops/utils.py

@@ -3,6 +3,7 @@
 #
 
 import functools
+import json
 import logging
 import os
 import re
@@ -383,6 +384,25 @@ def metadata(self) -> Optional[dict]:
             return None
         return yaml.safe_load((self.code_directory / METADATA_FILE_NAME).read_text())["data"]
 
+    @property
+    def connector_spec_file_content(self) -> Optional[dict]:
+        """
+        Returns spec file content from spec.yaml, spec.json, manifest.yaml.
+        This spec is not a "source of truth" for connector specification and can't be used as is.
+        """
+        yaml_spec = Path(self.python_source_dir_path / "spec.yaml")
+        json_spec = Path(self.python_source_dir_path / "spec.json")
+
+        if yaml_spec.exists():
+            return yaml.safe_load(yaml_spec.read_text())
+        elif json_spec.exists():
+            with open(json_spec) as f:
+                return json.load(f)
+        elif self.manifest_path.exists():
+            return yaml.safe_load(self.manifest_path.read_text())["spec"]
+
+        return None
+
     @property
     def language(self) -> ConnectorLanguage:
         if Path(self.code_directory / self.technical_name.replace("-", "_") / "manifest.yaml").is_file():

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/changelog.txt

@@ -0,0 +1,3 @@
+<details>
+  <summary>Expand to review</summary>
+</details>

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/for_airbyte_cloud.txt

@@ -0,0 +1,5 @@
+
+1. [Log into your Airbyte Cloud](https://cloud.airbyte.com/workspaces) account.
+2. Click Sources and then click + New source.
+3. On the Set up the source page, select {connector_name} from the Source type dropdown.
+4. Enter a name for the {connector_name} connector.

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/for_airbyte_open_source.txt

@@ -0,0 +1,2 @@
+
+1. Navigate to the Airbyte Open Source dashboard.

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/source.txt

@@ -0,0 +1,6 @@
+
+<HideInUI>
+
+This page contains the setup guide and reference information for the [{connector_name}]({docs_link}) source connector.
+
+</HideInUI>

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/supported_sync_modes.txt

@@ -0,0 +1,2 @@
+
+The {connector_name} source connector supports the following [sync modes](https://docs.airbyte.com/cloud/core-concepts/#connection-sync-modes):

airbyte-ci/connectors/connectors_qa/src/connectors_qa/checks/doc_templates/tutorials.txt

@@ -0,0 +1,2 @@
+
+Now that you have set up the {connector_name} source connector, check out the following {connector_name} tutorials:

airbyte-ci/connectors/connectors_qa/src/connectors_qa/models.py

@@ -123,6 +123,15 @@ def applies_to_connector_types(self) -> List[str]:
         """
         return ALL_TYPES
 
+    @property
+    def applies_to_connector_ab_internal_sl(self) -> int:
+        """The connector ab_internal_s that the QA check applies to
+
+        Returns:
+            int: integer value for connector ab_internal_sl level
+        """
+        return 0
+
     @property
     @abstractmethod
     def category(self) -> CheckCategory:
@@ -187,6 +196,11 @@ def run(self, connector: Connector) -> CheckResult:
                 connector,
                 f"Check does not apply to {connector.cloud_usage} connectors",
             )
+        if connector.ab_internal_sl < self.applies_to_connector_ab_internal_sl:
+            return self.skip(
+                connector,
+                f"Check does not apply to connectors with sl < {self.applies_to_connector_ab_internal_sl}",
+            )
         return self._run(connector)
 
     def _run(self, connector: Connector) -> CheckResult:

airbyte-ci/connectors/connectors_qa/src/connectors_qa/utils.py

@@ -1,10 +1,14 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 
+import re
+from difflib import get_close_matches
 from pathlib import Path
-from typing import Set
+from typing import Any, Set
 
 from connector_ops.utils import Connector  # type: ignore
 from connectors_qa import consts
+from markdown_it import MarkdownIt
+from markdown_it.tree import SyntaxTreeNode
 
 
 def remove_strict_encrypt_suffix(connector_technical_name: str) -> str:
@@ -42,3 +46,147 @@ def get_all_connectors_in_directory(directory: Path) -> Set[Connector]:
         if connector_directory.is_dir() and (connector_directory / consts.METADATA_FILE_NAME).exists():
             connectors.append(Connector(connector_directory.name))
     return set(connectors)
+
+
+def remove_step_from_heading(heading: str) -> str:
+    if "Step 1: " in heading:
+        return heading.replace("Step 1: ", "")
+    if "Step 2: " in heading:
+        return heading.replace("Step 2: ", "")
+    return heading
+
+
+def required_titles_from_spec(spec: dict[str, Any]) -> tuple[list[str], bool]:
+    has_credentials = False
+    spec_required = spec.get("required")
+    if not spec_required:
+        return [], False
+
+    spec_properties = spec["properties"].keys()
+    creds = ["credentials", "client_id", "client_secret", "access_token", "refresh_token", "authorization"]
+
+    if any(x in spec_required for x in creds):
+        has_credentials = True
+    if any(x in spec_properties for x in creds):
+        has_credentials = True
+    if has_credentials:
+        [spec_required.remove(cred) for cred in creds if cred in spec_required]
+
+    titles = [spec["properties"][field]["title"].lower() for field in spec_required]
+    return titles, has_credentials
+
+
+def documentation_node(connector_documentation: str) -> SyntaxTreeNode:
+    md = MarkdownIt("commonmark")
+    tokens = md.parse(connector_documentation)
+    return SyntaxTreeNode(tokens)
+
+
+def header_name(n: SyntaxTreeNode) -> str:
+    return n.to_tokens()[1].children[0].content
+
+
+def prepare_lines_to_compare(connector_name: str, docs_line: str, template_line: str) -> tuple[str]:
+    def _replace_link(docs_string: str, link_to_replace: str) -> str:
+        links = re.findall("(https?://[^\s)]+)", docs_string)
+        for link in links:
+            docs_string = docs_string.replace(link, link_to_replace)
+        return docs_string
+
+    connector_name_to_replace = "{connector_name}"
+    link_to_replace = "{docs_link}"
+
+    template_line = (
+        template_line.replace(connector_name_to_replace, connector_name) if connector_name_to_replace in template_line else template_line
+    )
+    docs_line = _replace_link(docs_line, link_to_replace) if link_to_replace in template_line else docs_line
+
+    return docs_line, template_line
+
+
+def remove_not_required_step_headers(headers: tuple[str]) -> tuple[str]:
+    """
+    Removes headers like Step 1.1 Step 3 Step 2.3 from actual headers, if they placed after Step 1: header.
+    from: "Connector name", "Prerequisites", "Setup guide", "Step 1: do something 1", "Step 1.11: do something 11",
+           "Step 2: do something 2", "Step 2.1: do something 2.1", "Changelog"
+    To: "Connector name", "Prerequisites", "Setup guide", "Step 1: do something 1", "Step 2: do something 2", "Changelog"
+    This is connector specific headers, so we can ignore them.
+    """
+    step_one_index = None
+    for header in headers:
+        if re.search("Step 1: ", header):
+            step_one_index = headers.index(header)
+    if not step_one_index:  # docs doesn't have Step 1 headers
+        return headers
+
+    step_headers = headers[step_one_index:]
+    pattern = "Step \d+.?\d*: "
+    step = "Step 1: "
+    i = 0
+    while i < len(step_headers):
+        if step in step_headers[i]:  # if Step 1/2: is substring of current header
+            if i + 1 < len(step_headers) and re.match(pattern, step_headers[i + 1]):  # check that header has Step x:
+                if "Step 2: " in step_headers[i + 1]:  # found Step 2, it's required header, move to the next one
+                    step = "Step 2: "
+                    i += 1
+                    continue
+                else:
+                    step_headers.remove(step_headers[i + 1])  # remove all other steps from headers
+                    continue  # move to the next header after Step 1/2 header
+            else:
+                break
+        break
+
+    headers = headers[:step_one_index]
+    headers.extend(step_headers)
+    return headers
+
+
+def reason_titles_not_match(heading_names_value: str, template_headings_value: str, template_headings: list[str]) -> str:
+    reason = f"Heading '{heading_names_value}' is not in the right place, the name of heading is incorrect or not expected.\n"
+    close_titles = get_close_matches(heading_names_value, template_headings)
+    if close_titles and close_titles[0] != heading_names_value:
+        diff = f"Diff:\nActual Heading: '{heading_names_value}'. Possible correct heading: '{close_titles}'. Expected Heading: '{template_headings_value}'"
+    else:
+        diff = f"Diff:\nActual Heading: '{heading_names_value}'. Expected Heading: '{template_headings_value}'"
+    return reason + diff
+
+
+def reason_missing_titles(template_headings_index: int, template_headings: list[str], not_required_headers: list[str]) -> str:
+    missing = template_headings[template_headings_index:]
+    required = [m for m in missing if m not in not_required_headers]
+    return f"Required missing headers: {required}. All missing headers: {missing}"
+
+
+def description_end_line_index(heading: str, actual_headings: list[str], header_line_map: dict[str, int]) -> int:
+    if actual_headings.index(heading) + 1 == len(actual_headings):
+        return
+    return header_line_map[actual_headings[actual_headings.index(heading) + 1]]
+
+
+def prepare_headers(connector_documentation: dict) -> list[str]:
+    node = documentation_node(connector_documentation)
+    headers = [header_name(n) for n in node if n.type == "heading"]  # find all headers
+    headers = remove_not_required_step_headers(headers)  # remove Step 1.1 Step 3 ... headers
+    headers = tuple([remove_step_from_heading(h) for h in headers])  # remove Step 1 and Step 2 from header name
+    return headers
+
+
+def prepare_changelog_to_compare(docs: list[str]) -> list[str]:
+    docs_to_compare = []
+    _siblings_content = []
+    n = "\n"
+    docs = "".join(docs)
+    node = documentation_node(docs)
+
+    for sibling in node[0].siblings:
+        _siblings_content.append(sibling.content.rstrip())
+
+    for c in _siblings_content:
+        if n in c:
+            docs_to_compare += [_c + n for _c in c.split(n)]
+        else:
+            docs_to_compare.append(c)
+
+    docs_to_compare = list(filter(("").__ne__, docs_to_compare))
+    return docs_to_compare