# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import pathlib
from typing import Any, Optional

from typing_extensions import override

from crossbench.action_runner.base import ActionRunner
from crossbench.action_runner.default_action_runner import DefaultActionRunner
from crossbench.probes.probe import Probe
from crossbench.probes.results import (BrowserProbeResult,
                                       DuplicateProbeResult, EmptyProbeResult,
                                       LocalProbeResult, ProbeResultDict)
from tests import test_helper
from tests.crossbench.base import (BaseCrossbenchTestCase,
                                   CrossbenchFakeFsTestCase)


class ProbeResultTestCase(CrossbenchFakeFsTestCase):

  def test_is_empty(self):
    empty = EmptyProbeResult()
    self.assertTrue(empty.is_empty)
    self.assertFalse(empty)
    combined = empty.merge(EmptyProbeResult())
    self.assertTrue(combined.is_empty)
    url = LocalProbeResult(url=("http://test.com",))
    self.assertFalse(url.is_empty)
    self.assertTrue(url)
    combined = empty.merge(url)
    self.assertFalse(combined.is_empty)
    self.assertTrue(combined)

  def test_equal_empty(self):
    empty = EmptyProbeResult()
    self.assertEqual(empty, empty)
    self.assertEqual(EmptyProbeResult(), EmptyProbeResult())
    local_empty = LocalProbeResult()
    self.assertEqual(local_empty, EmptyProbeResult())
    self.assertEqual(local_empty, local_empty)
    self.assertNotEqual(LocalProbeResult(), None)
    self.assertNotEqual(None, LocalProbeResult())

  def test_is_remote(self):
    empty = EmptyProbeResult()
    self.assertFalse(empty.is_remote)
    local_empty = LocalProbeResult()
    self.assertFalse(local_empty.is_remote)

  def test_equal_single(self):
    url = "http://test.com"
    self.assertEqual(LocalProbeResult(url=(url,)), LocalProbeResult(url=(url,)))
    url_b = "http://foo.test.com"
    self.assertNotEqual(
        LocalProbeResult(url=(url,)), LocalProbeResult(url=(url_b,)))

  def test_invalid_files(self):
    with self.assertRaises(ValueError):
      LocalProbeResult(file=[pathlib.Path("foo.json")])
    with self.assertRaises(ValueError):
      LocalProbeResult(file=[pathlib.Path("foo.csv")])
    with self.assertRaises(ValueError):
      LocalProbeResult(json=[pathlib.Path("foo.csv")])
    with self.assertRaises(ValueError):
      LocalProbeResult(csv=[pathlib.Path("foo.json")])

  def test_inexistent_files(self):
    with self.assertRaises(ValueError):
      LocalProbeResult(file=[pathlib.Path("not_there.txt")])
    with self.assertRaises(ValueError):
      LocalProbeResult(csv=[pathlib.Path("not_there.csv")])
    with self.assertRaises(ValueError):
      LocalProbeResult(json=[pathlib.Path("not_there.json")])

  def test_result_url(self):
    url = "http://foo.bar.com"
    with self.assertRaises(DuplicateProbeResult):
      _ = LocalProbeResult(url=[url, url])
    result = LocalProbeResult(url=[url])
    self.assertFalse(result.is_empty)
    self.assertEqual(result.url, url)
    self.assertListEqual(result.url_list, [url])
    self.assertListEqual(list(result.all_files()), [])
    failed: Any = None
    with self.assertRaises(ValueError):
      failed = result.file
    with self.assertRaises(ValueError):
      failed = result.json
    with self.assertRaises(ValueError):
      failed = result.csv
    self.assertIsNone(failed)
    json_data = result.to_json()
    self.assertDictEqual(json_data, {"url": (url,)})

  def test_result_any_file(self):
    path = self.create_file("result.txt")
    with self.assertRaises(DuplicateProbeResult):
      _ = LocalProbeResult(file=[path, path])
    result = LocalProbeResult(file=[path])
    self.assertFalse(result.is_empty)
    self.assertNotEqual(
        result, LocalProbeResult(file=[
            self.create_file("result2.txt"),
        ]))
    self.assertEqual(result.file, path)
    self.assertListEqual(result.file_list, [path])
    self.assertListEqual(list(result.all_files()), [path])
    failed: Any = None
    with self.assertRaises(ValueError):
      failed = result.url
    with self.assertRaises(ValueError):
      failed = result.json
    with self.assertRaises(ValueError):
      failed = result.csv
    self.assertIsNone(failed)

  def test_result_csv(self):
    path = self.create_file("result.csv")
    with self.assertRaises(DuplicateProbeResult):
      _ = LocalProbeResult(csv=[path, path])
    result = LocalProbeResult(csv=[path])
    self.assertFalse(result.is_empty)
    self.assertNotEqual(
        result, LocalProbeResult(csv=[
            self.create_file("result2.csv"),
        ]))
    self.assertEqual(result.csv, path)
    self.assertEqual(result.get("csv"), path)
    self.assertListEqual(result.csv_list, [path])
    self.assertListEqual(result.get_all("csv"), [path])
    self.assertListEqual(list(result.all_files()), [path])
    self.assertEqual(result.file, path)
    failed: Any = None
    with self.assertRaises(ValueError):
      failed = result.url
    with self.assertRaises(ValueError):
      failed = result.json
    self.assertIsNone(failed)

  def test_result_json(self):
    path = self.create_file("result.json")
    with self.assertRaises(DuplicateProbeResult):
      _ = LocalProbeResult(json=[path, path])
    with self.assertRaises(DuplicateProbeResult):
      _ = LocalProbeResult(file=[path, path])
    result = LocalProbeResult(json=[path])
    self.assertFalse(result.is_empty)
    self.assertNotEqual(
        result, LocalProbeResult(json=[
            self.create_file("result2.json"),
        ]))
    self.assertEqual(result.json, path)
    self.assertEqual(result.get("json"), path)
    self.assertListEqual(result.json_list, [path])
    self.assertListEqual(result.get_all("json"), [path])
    self.assertListEqual(list(result.all_files()), [path])
    self.assertEqual(result.file, path)
    failed: Any = None
    with self.assertRaises(ValueError):
      failed = result.url
    with self.assertRaises(ValueError):
      failed = result.csv
    self.assertIsNone(failed)

  def test_multiple_urls(self):
    url1 = "http://one.com"
    url2 = "http://two.com"
    result = LocalProbeResult(url=(url1, url2))
    self.assertFalse(result.is_empty)
    self.assertFalse(result.has_files)
    with self.assertRaises(ValueError):
      _ = result.file
    with self.assertRaises(ValueError):
      _ = result.url
    self.assertSequenceEqual(result.url_list, (url1, url2))

  def test_multiple_files(self):
    json1 = self.create_file("result_1.json")
    json2 = self.create_file("result_2.json")
    zip1 = self.create_file("result.zip")
    result = LocalProbeResult(file=(json1, json2, zip1))
    self.assertFalse(result.is_empty)
    self.assertTrue(result.has_files)
    with self.assertRaises(ValueError):
      _ = result.file
    with self.assertRaises(ValueError):
      _ = result.json
    self.assertSequenceEqual(result.file_list, (json1, json2, zip1))
    self.assertSequenceEqual(result.json_list, (json1, json2))
    self.assertSequenceEqual(result.get_all("json"), (json1, json2))
    self.assertEqual(result.get("zip"), zip1)
    self.assertEqual(result.get_all("zip"), [zip1])
    with self.assertRaises(ValueError):
      _ = result.get("other")

  def test_merge(self):
    file = self.create_file("result.custom")
    json = self.create_file("result.json")
    csv = self.create_file("result.csv")
    url = "http://foo.bar.com"
    trace = self.create_file("trace.pb")

    result = LocalProbeResult(
        url=(url,), file=(file,), json=(json,), csv=(csv,), trace=(trace,))
    self.assertFalse(result.is_empty)
    self.assertListEqual(list(result.all_files()), [file, json, csv, trace])
    self.assertListEqual(result.url_list, [url])
    self.assertListEqual(result.trace_list, [trace])

    merged = result.merge(EmptyProbeResult())
    self.assertFalse(merged.is_empty)
    self.assertListEqual(list(merged.all_files()), [file, json, csv, trace])
    self.assertListEqual(merged.url_list, [url])
    self.assertListEqual(merged.trace_list, [trace])

    file_2 = self.create_file("result.2.custom")
    json_2 = self.create_file("result.2.json")
    csv_2 = self.create_file("result.2.csv")
    url_2 = "http://foo.bar.com/2"
    trace_2 = self.create_file("trace.2.pb")
    other = LocalProbeResult(
        url=(url_2,),
        file=(file_2,),
        json=(json_2,),
        csv=(csv_2,),
        trace=(trace_2,))
    merged = result.merge(other)
    self.assertFalse(merged.is_empty)
    self.assertListEqual(
        list(merged.all_files()),
        [file, file_2, json, json_2, csv, csv_2, trace, trace_2])
    self.assertListEqual(merged.url_list, [url, url_2])
    # result is unchanged:
    self.assertFalse(result.is_empty)
    self.assertListEqual(list(result.all_files()), [file, json, csv, trace])
    self.assertListEqual(result.url_list, [url])
    self.assertListEqual(result.trace_list, [trace])
    # other is unchanged:
    self.assertFalse(other.is_empty)
    self.assertListEqual(
        list(other.all_files()), [file_2, json_2, csv_2, trace_2])
    self.assertListEqual(other.url_list, [url_2])
    self.assertListEqual(other.trace_list, [trace_2])

  def test_merge_duplicate_files(self):
    path = self.create_file("result.custom")
    result_1 = LocalProbeResult(file=(path,))
    result_2 = LocalProbeResult(file=(path,))
    with self.assertRaises(DuplicateProbeResult):
      result_1.merge(result_1)
    with self.assertRaises(DuplicateProbeResult):
      result_1.merge(result_2)
    with self.assertRaises(DuplicateProbeResult):
      result_2.merge(result_1)

  def test_traces_are_files(self):
    trace = self.create_file("trace.pb")

    result = LocalProbeResult(trace=(trace,))
    self.assertFalse(result.is_empty)
    self.assertListEqual(list(result.all_files()), [trace])
    self.assertListEqual(list(result.trace_list), [trace])

  def test_traces_can_be_duplicate(self):
    trace = self.create_file("trace.pb")

    result = LocalProbeResult(trace=(trace,), file=(trace,))
    self.assertFalse(result.is_empty)
    self.assertListEqual(list(result.all_files()), [trace])
    self.assertListEqual(list(result.trace_list), [trace])


class MockRun:

  def __init__(self,
               browser,
               browser_platform,
               action_runner: Optional[ActionRunner] = None):
    self.browser = browser
    self.browser_platform = browser_platform
    self.is_remote = False
    self.action_runner: ActionRunner = action_runner or DefaultActionRunner()


class BrowserProbeResultTestCase(BaseCrossbenchTestCase):

  @override
  def setUp(self) -> None:
    super().setUp()
    self.run = MockRun(self.browsers[0], self.platform)

  def test_empty(self):
    result = BrowserProbeResult(self.run)
    self.assertTrue(result.is_empty)
    self.assertFalse(result.is_remote)

  def test_remote_empty(self):
    self.run.is_remote = True
    result = BrowserProbeResult(self.run)
    self.assertTrue(result.is_empty)
    self.assertEqual(result, EmptyProbeResult())
    self.assertEqual(result, LocalProbeResult())

  def test_remote_only_urls(self):
    self.run.is_remote = True
    result = BrowserProbeResult(self.run, url=("http://test.com",))
    self.assertFalse(result.is_empty)
    self.assertNotEqual(result, EmptyProbeResult())
    self.assertEqual(result, LocalProbeResult(url=("http://test.com",)))
    self.assertNotEqual(result, LocalProbeResult(url=("http://test.com/bar",)))

  def test_local_browser_equal_local_run(self):
    url = "http://foo.bar.com"
    file = self.create_file("results/file.any")
    json = self.create_file("results/file.json")
    csv = self.create_file("results/file.csv")
    local = LocalProbeResult((url,), (file, json, csv))
    browser = BrowserProbeResult(self.run, (url,), (file, json, csv))
    self.assertEqual(local, browser)
    local_kwargs = LocalProbeResult(
        url=(url,), file=(file,), json=(json,), csv=(csv,))
    browser_kwargs = BrowserProbeResult(
        self.run, url=(url,), file=(file,), json=(json,), csv=(csv,))
    self.assertEqual(local, browser)
    self.assertEqual(local, browser_kwargs)
    self.assertEqual(local_kwargs, browser)
    self.assertEqual(local_kwargs, browser_kwargs)

  def test_copy_remote_files(self):
    # pylint: disable=attribute-defined-outside-init
    out_dir_local = pathlib.Path("local/results")
    out_dir_remote = pathlib.Path("remote/results")
    out_dir_local.mkdir(parents=True)
    out_dir_remote.mkdir(parents=True)
    self.assertTrue(out_dir_local.is_dir())
    self.assertTrue(out_dir_remote.is_dir())
    remote_txt = self.create_file(
        out_dir_remote / "result.txt", contents="a1 b2 c3")
    remote_json = self.create_file(
        out_dir_remote / "result.json", contents="[]")
    self.run.is_remote = True
    self.run.out_dir = out_dir_local
    self.run.browser_tmp_dir = out_dir_remote
    with self.assertRaises(DuplicateProbeResult):
      _ = BrowserProbeResult(self.run, file=(remote_txt, remote_txt))
    with self.assertRaises(DuplicateProbeResult):
      _ = BrowserProbeResult(self.run, json=(remote_json, remote_json))
    result = BrowserProbeResult(
        self.run, file=(remote_txt,), json=(remote_json,))
    self.assertTrue(remote_txt.is_file())
    with self.assertRaises(ValueError):
      _ = result.file
    self.assertTrue(result.has_files)
    self.assertNotEqual(remote_json, result.json)
    self.assertTrue(result.json.is_file())
    self.assertEqual(result.json.read_text(), "[]")
    self.assertNotEqual(remote_txt, result.get("txt"))
    self.assertEqual(result.get("txt").read_text(), "a1 b2 c3")


class MockProbe(Probe):
  """
  Probe DOC Text
  """
  NAME = "mock-probe"

  @override
  def get_context_cls(self):
    pass


class ProbeResultDictTestCase(CrossbenchFakeFsTestCase):

  @override
  def setUp(self) -> None:
    super().setUp()
    self.result_location = pathlib.Path("test/out/results")
    self.result_dict = ProbeResultDict(self.result_location)
    self.local_result = LocalProbeResult()

  def test_create_empty(self):
    self.assertEqual(len(self.result_dict), 0)
    self.assertFalse(self.result_dict)
    self.assertNotIn(MockProbe(), self.result_dict)

  def test_get_item(self):
    probe = MockProbe()
    with self.assertRaises(KeyError):
      _ = self.result_dict[probe]
    self.result_dict[probe] = self.local_result
    self.assertIs(self.result_dict[probe], self.local_result)

  def test_get(self):
    probe = MockProbe()
    self.assertIsNone(self.result_dict.get(probe))
    self.assertEqual(self.result_dict.get(probe, 1234), 1234)
    self.result_dict[probe] = self.local_result
    self.assertIs(self.result_dict.get(probe), self.local_result)

  def test_to_json_empty(self):
    probe = MockProbe()
    self.result_dict[probe] = self.local_result
    json = self.result_dict.to_json()
    self.assertDictEqual(json, {MockProbe.NAME: None})

  def test_to_json_result(self):
    csv = self.create_file("result.csv")
    txt = self.create_file("result.txt")
    self.local_result = LocalProbeResult(csv=(csv,), txt=(txt,))
    probe = MockProbe()
    self.result_dict[probe] = self.local_result
    json = self.result_dict.to_json()
    self.assertDictEqual(
        json, {MockProbe.NAME: {
            "csv": [str(csv)],
            "txt": [str(txt)]
        }})


if __name__ == "__main__":
  test_helper.run_pytest(__file__)