text: from __future__ import annotations
  
  import P4
  
  import p4transfer
  
  
  def test_streams_basic(
      source,
      target,
      transfer_client,
      default_transfer_config,
  ):
      """Test basic source/target being a stream."""
      d = source.fetch("depot", "src_streams")
      d["Type"] = "stream"
      source.save("depot", d)
      s = source.fetch("stream", "-t", "mainline", "//src_streams/main")
      source.save("stream", s)
      d = target.fetch("depot", "targ_streams")
      d["Type"] = "stream"
      target.save("depot", d)
      config = default_transfer_config.copy()
      config["views"] = ""
      config["transfer_target_stream"] = "//targ_streams/transfer_target_stream"
      config["stream_views"] = [
          {
              "src": "//src_streams/main",
              "targ": "//targ_streams/main",
              "stype": "mainline",
              "parent": "",
          }
      ]
      c = source.fetch("client", source.client_name)
      c["Stream"] = "//src_streams/main"
      source.save("client", c)
      file1 = source.local_path("inside/file1")
      file1.write_bytes(b"Test content")
      source.p4("add", file1)
      source.p4("submit", "-d", "Added files")
      p4transfer.test_transfer(config)
      # You might expect one change, but there are extras due to stream creation.
      assert target.counter == 2
      assert len(target.p4("changes")) == 3
      changes = target.p4("changes", "//targ_streams/...")
      assert len(changes) == 1
      filelog = target.filelog("//targ_streams/main/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "add"
      exp_view = [f"//src_streams/main/... //{transfer_client}/src_streams/main/..."]
      client = source.fetch("client", transfer_client)
      assert exp_view == client._view
      client = target.fetch("client", transfer_client)
      assert client._stream == "//targ_streams/transfer_target_stream"
      # Run again and expect no extra stream changes to be created.
      p4transfer.test_transfer(config)
      assert target.counter == 2
      assert len(target.p4("changes")) == 3
  
  
  def test_streams_multiple(
      source,
      target,
      default_transfer_config,
  ):
      """Test source/target being streams with multiples."""
      d = source.fetch("depot", "src_streams")
      d["Type"] = "stream"
      source.save("depot", d)
      s = source.fetch("stream", "-t", "mainline", "//src_streams/main")
      source.save("stream", s)
      d = target.fetch("depot", "targ_streams")
      d["Type"] = "stream"
      target.save("depot", d)
      s = target.fetch("stream", "-t", "mainline", "//targ_streams/main")
      target.save("stream", s)
      config = default_transfer_config.copy()
      config["views"] = ""
      config["transfer_target_stream"] = "//targ_streams/transfer_target_stream"
      config["stream_views"] = [
          {
              "src": "//src_streams/rel*",
              "targ": "//targ_streams/rel*",
              "stype": "release",
              "parent": "//targ_streams/main",
          }
      ]
      c = source.fetch("client", source.client_name)
      c["Stream"] = "//src_streams/main"
      source.save("client", c)
      file1 = source.local_path("inside/file1")
      file1.write_bytes(b"Test content")
      source.p4("add", file1)
      source.p4("submit", "-d", "Added files")
      try:
          p4transfer.test_transfer(config)
      except P4.P4Exception as e:
          assert "no such stream" in str(e)
      assert target.counter == 0
      assert len(target.p4("changes")) == 1
      s = source.fetch(
          "stream", "-t", "release", "-P", "//src_streams/main", "//src_streams/rel1"
      )
      source.save("stream", s)
      s = source.fetch(
          "stream", "-t", "release", "-P", "//src_streams/main", "//src_streams/rel2"
      )
      source.save("stream", s)
      source.p4("populate", "-S", "//src_streams/rel1", "-r")
      source.p4("populate", "-S", "//src_streams/rel2", "-r")
      p4transfer.test_transfer(config)
      changes = target.p4("changes", "//targ_streams/...")
      assert len(changes) == 2
      filelog = target.filelog("//targ_streams/rel1/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "add"
      filelog = target.filelog("//targ_streams/rel2/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "add"
      # Use double wildcards.
      config["stream_views"] = [
          {
              "src": "//src_*/rel*",
              "targ": "//targ_*/rel*",
              "stype": "release",
              "parent": "//targ_streams/main",
          }
      ]
      c = source.fetch("client", source.client_name)
      c["Stream"] = "//src_streams/rel1"
      source.save("client", c)
      file2 = source.local_path("inside/file2")
      file2.write_bytes(b"Test content")
      source.p4("add", file2)
      source.p4("submit", "-d", "Added file2")
      p4transfer.test_transfer(config)
      changes = target.p4("changes", "//targ_streams/...")
      assert len(changes) == 3
      filelog = target.filelog("//targ_streams/rel1/...")
      assert len(filelog) == 2
      assert filelog[0].revisions[0].action == "add"
      assert filelog[1].revisions[0].action == "add"
      filelog = target.filelog("//targ_streams/rel2/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "add"
      # Now change source streams, and replicate again - expecting new stream to be picked up.
      s = source.fetch(
          "stream", "-t", "release", "-P", "//src_streams/main", "//src_streams/rel3"
      )
      source.save("stream", s)
      source.p4("populate", "-S", "//src_streams/rel3", "-r")
      p4transfer.test_transfer(config)
      assert target.counter == 9
      assert len(target.p4("changes")) == 10
      changes = target.p4("changes", "//targ_streams/...")
      assert len(changes) == 4
      filelog = target.filelog("//targ_streams/rel3/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "add"
  
  
  def test_streams_multiple_individual(
      source,
      target,
      default_transfer_config,
  ):
      """Test source/target being streams with multiple source/target matching."""
      d = source.fetch("depot", "src_streams")
      d["Type"] = "stream"
      source.save("depot", d)
      s = source.fetch("stream", "-t", "mainline", "//src_streams/main")
      source.save("stream", s)
      d = target.fetch("depot", "targ_streams")
      d["Type"] = "stream"
      target.save("depot", d)
      s = target.fetch("stream", "-t", "mainline", "//targ_streams/main")
      target.save("stream", s)
      config = default_transfer_config.copy()
      config["views"] = ""
      config["transfer_target_stream"] = "//targ_streams/transfer_target_stream"
      config["stream_views"] = [
          {
              "src": "//src_streams/main",
              "targ": "//targ_streams/main",
              "stype": "mainline",
              "parent": "",
          },
          {
              "src": "//src_streams/rel1",
              "targ": "//targ_streams/rel1",
              "stype": "release",
              "parent": "//targ_streams/main",
          },
          {
              "src": "//src_streams/rel2",
              "targ": "//targ_streams/rel2",
              "stype": "release",
              "parent": "//targ_streams/main",
          },
      ]
      c = source.fetch("client", source.client_name)
      c["Stream"] = "//src_streams/main"
      source.save("client", c)
      file1 = source.local_path("inside/file1")
      file1.write_bytes(b"Test content")
      source.p4("add", file1)
      source.p4("submit", "-d", "Added files")
      s = source.fetch(
          "stream", "-t", "release", "-P", "//src_streams/main", "//src_streams/rel1"
      )
      source.save("stream", s)
      s = source.fetch(
          "stream", "-t", "release", "-P", "//src_streams/main", "//src_streams/rel2"
      )
      source.save("stream", s)
      source.p4("populate", "-S", "//src_streams/rel1", "-r")
      source.p4("populate", "-S", "//src_streams/rel2", "-r")
      p4transfer.test_transfer(config)
      assert target.counter == 6
      assert len(target.p4("changes")) == 7
      changes = target.p4("changes", "//targ_streams/...")
      assert len(changes) == 3
      filelog = target.filelog("//targ_streams/rel1/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "branch"
      filelog = target.filelog("//targ_streams/rel2/...")
      assert len(filelog) == 1
      assert filelog[0].revisions[0].action == "branch"
  