test_p4review2.py #1

#!/usr/bin/env python3
# -*- encoding: UTF8 -*-
# Test harness for p4review2.py

import sys
import unittest
import os
import subprocess

import P4
curr_dir = os.path.dirname(os.path.abspath(__file__))

# ../../../../../../Unsupported/Samples/triggers/tests

sys.path.insert(0, os.path.join(curr_dir, "..", "..", "..", "..", "..", "..", "Unsupported", "Samples", "triggers", "tests"))

from p4testutils import TestCase, P4Server, localDirectory, create_file

# from p4review2 import P4Review

os.environ["LOGS"] = "."
LOGGER_NAME = "Testp4review2"
LOG_FILE = "log-Testp4review2.log"


class Testp4review2(TestCase):
    def __init__(self, methodName='runTest'):
        super(Testp4review2, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName)

    def setUp(self):
        self.server = P4Server()
        p4 = self.server.p4
        self.p4 = p4
        p4.logger = self.logger

    def tearDown(self):
        pass

    def testP4review2(self):
        """check that fixes are appropriate controlled"""

        p4 = self.p4

        config = """
[p4review]
change_template = Change {chgno} by {p4user}@{p4client} on {dt}
	{change_url}
	{user_url}
	{cldesc}
	 .
	Jobs updated:
	{jobsupdated}
	 .
	Affected files:
	{clfiles}

change_url = http://p4web:1680/{chgno}?ac=10
daemon =
dbfile = :memory:
default_domain = example.org
default_sender = Perforce Review Daemon <perforce-review-daemon>
html_change_template =
	<div style="font-family: sans-serif;">
	Change <a style="text-decoration: none;" href="{change_url}">{chgno}</a>
	by <a style="text-decoration: none;" href="{user_url}">{p4user}</a>@{p4client}
	on {dt}
	<br/>
	<div style="margin: 1em;">{cldesc}</div>
	<br/>
	Jobs updated:
	<ul style="margin: 1em; padding: 0; list-style-type: none;">
	{jobsupdated}
	</ul>
	<br/>
	Affected files:
	<ul style="margin-left: 1em; padding: 0; list-style-type: none;">
	{clfiles}
	</ul>
	</div>

html_files_template = <li style="margin:0; padding:0;"><a style="text-decoration: none;" href="{change_url}#{fhash}">{dfile}</a>#{drev} {action}</li>
html_job_template =
	<a href="{job_url}">{Job}</a>
	<dl>
	{jobdesc}
	</dl>
ignored_users = git-fusion-reviews-*
job_counter =
job_datefield = Date
job_template = {job_url}
	{jobdesc}

job_url = http://p4web:1680/{jobno}?ac=111
log_file =
max_email_size = 1048576
max_emails = 99
max_length = 4096
opt_in_path =
p4bin = /p4/common/bin/p4
p4charset = utf8
p4passwd =
p4port = 1666
p4user = rcowham
pid_file = /tmp/p4review2.pid
poll_interval = 1
review_counter = review
skip_author = True
smtp_passwd =
smtp_server = smtp:25
smtp_ssl = NONE
smtp_user = p4review
spec_depot = spec
subject_template = [{p4port} @{chgno}] {desc}
summary_email = False
timeoffset = 0.0
user_url = http://p4web:1680/{p4user}?ac=17
"""

        files = localDirectory(self.server.client_root, "files")
        d1 = localDirectory(files, "d1")
        d2 = localDirectory(files, "d2")
        for i in range(10):
            f = os.path.join(d1, "file_%d" % i)
            create_file(f, "0123456789\n" * 10)
            f = os.path.join(d2, "file_%d" % i)
            create_file(f, "0123456789\n" * 10)
        p4.run_add("%s/..." % files)
        p4.run_submit("-d", "Initial change")

        user = p4.fetch_user("fred")
        user["Reviews"] = ["//depot/files/d1/..."]
        p4.save_user(user, "-f")

        config_file = os.path.join(self.server.client_root, "review.cfg")
        with open(config_file, "w") as f:
            f.write(config)

        try:
            shell_script = f'{os.path.join(curr_dir, "..", "p4review2.py")} -c {config_file} -p "{self.p4.port}" -u {self.p4.user} -f --debug-email -L /tmp/p4review2.log'
            result = subprocess.run(
                shell_script, 
                shell=True,
                check=True,
                stdout=subprocess.PIPE,  # Capture standard output
                stderr=subprocess.PIPE   # Capture standard error
            )
            
            output = result.stdout.decode('utf-8')  # Decode the output from bytes to string
            error_output = result.stderr.decode('utf-8')  # Decode the error output
            
#             From: Perforce Review Daemon <perforce-review-daemon@example.org>
# Reply-To: testuser <testuser@test_ws>
# To: fred <fred@test_ws>
# Subject: [rsh:p4d -r /Users/rcowham/vagrant/sdpdev/sdp/Server/Unix/p4/common/bin/tes...

            # Parse the outputs
            self.assertRegex(output, r'(?m)From: Perforce Review Daemon')
            self.assertRegex(output, r'(?m)Reply-To: testuser')
            self.assertRegex(output, r'(?m)To: fred')
            self.assertRegex(output, r'(?m)Subject: \[rsh:p4d')
            # print("Output:", output)
            if error_output:
                print("Error:", error_output)

        except subprocess.CalledProcessError as e:
            print(f"Error executing the script: {e}")
            print(f"Error Output: {e.stderr.decode('utf-8')}")

if __name__ == '__main__':
    unittest.main()