#!/usr/bin/env vpython3
# Copyright 2020 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import logging
import os
import re
import sys

ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GCLIENT_PATH = os.path.join(ROOT_DIR, "gclient")
sys.path.insert(0, ROOT_DIR)

import subprocess2  # noqa: E402
from testing_support import fake_repos  # noqa: E402


class GClientSmokeBase(fake_repos.FakeReposTestBase):
    def setUp(self):
        super(GClientSmokeBase, self).setUp()
        # Make sure it doesn't try to auto update when testing!
        self.env = os.environ.copy()
        self.env["DEPOT_TOOLS_UPDATE"] = "0"
        self.env["DEPOT_TOOLS_METRICS"] = "0"
        # Suppress Python 3 warnings and other test undesirables.
        self.env["GCLIENT_TEST"] = "1"
        self.maxDiff = None

    def gclient(self, cmd, cwd=None, error_ok=False):
        if not cwd:
            cwd = self.root_dir
        cmd = [GCLIENT_PATH] + cmd
        process = subprocess2.Popen(
            cmd,
            cwd=cwd,
            env=self.env,
            stdout=subprocess2.PIPE,
            stderr=subprocess2.PIPE,
            universal_newlines=True,
        )
        (stdout, stderr) = process.communicate()
        logging.debug("XXX: %s\n%s\nXXX" % (" ".join(cmd), stdout))
        logging.debug("YYY: %s\n%s\nYYY" % (" ".join(cmd), stderr))

        if not error_ok:
            self.assertEqual(0, process.returncode, stderr)

        return (
            stdout.replace("\r\n", "\n"),
            stderr.replace("\r\n", "\n"),
            process.returncode,
        )

    def untangle(self, stdout):
        """Separates output based on thread IDs."""
        tasks = {}
        remaining = []
        task_id = 0
        for line in stdout.splitlines(False):
            m = re.match(r"^(\d)+>(.*)$", line)
            if not m:
                if task_id:
                    # Lines broken with carriage breaks don't have a thread ID,
                    # but belong to the last seen thread ID.
                    tasks.setdefault(task_id, []).append(line)
                else:
                    remaining.append(line)
            else:
                self.assertEqual([], remaining)
                task_id = int(m.group(1))
                tasks.setdefault(task_id, []).append(m.group(2))
        out = []
        for key in sorted(tasks.keys()):
            out.extend(tasks[key])
        out.extend(remaining)
        return "\n".join(out)

    def parseGclient(self, cmd, items, expected_stderr="", untangle=False):
        """Parse gclient's output to make it easier to test.
        If untangle is True, tries to sort out the output from parallel checkout."""
        (stdout, stderr, _) = self.gclient(cmd)
        if untangle:
            stdout = self.untangle(stdout)
        self.checkString(expected_stderr, stderr)
        return self.checkBlock(stdout, items)

    def splitBlock(self, stdout):
        """Split gclient's output into logical execution blocks.
        ___ running 'foo' at '/bar'
        (...)
        ___ running 'baz' at '/bar'
        (...)

        will result in 2 items of len((...).splitlines()) each.
        """
        results = []
        for line in stdout.splitlines(False):
            # Intentionally skips empty lines.
            if not line:
                continue
            if not line.startswith("__"):
                if results:
                    results[-1].append(line)
                else:
                    # TODO(maruel): gclient's git stdout is inconsistent.
                    # This should fail the test instead!!
                    pass
                continue

            match = re.match(r"^________ ([a-z]+) \'(.*)\' in \'(.*)\'$", line)
            if match:
                results.append(
                    [[match.group(1), match.group(2), match.group(3)]]
                )
                continue

            match = re.match(r"^_____ (.*) is missing, syncing instead$", line)
            if match:
                # Blah, it's when a dependency is deleted, we should probably
                # not output this message.
                results.append([line])
                continue

            # These two regexps are a bit too broad, they are necessary only for
            # git checkouts.
            if re.match(r"_____ [^ ]+ at [^ ]+", line) or re.match(
                r"_____ [^ ]+ : Attempting rebase onto [0-9a-f]+...", line
            ):
                continue

            # Fail for any unrecognized lines that start with '__'.
            self.fail(line)
        return results

    def checkBlock(self, stdout, items):
        results = self.splitBlock(stdout)
        for i in range(min(len(results), len(items))):
            if isinstance(items[i], (list, tuple)):
                verb = items[i][0]
                path = items[i][1]
            else:
                verb = items[i]
                path = self.root_dir
            self.checkString(
                results[i][0][0], verb, (i, results[i][0][0], verb)
            )
            if sys.platform == "win32":
                # Make path lower case since casing can change randomly.
                self.checkString(
                    results[i][0][2].lower(),
                    path.lower(),
                    (i, results[i][0][2].lower(), path.lower()),
                )
            else:
                self.checkString(
                    results[i][0][2], path, (i, results[i][0][2], path)
                )
        self.assertEqual(
            len(results), len(items), (stdout, items, len(results))
        )
        return results