"""End-to-end tests. Run PHP_BIN=php python3 tests/run.py. Uses an isolated copy."""
import base64, hashlib, hmac, http.cookiejar, json, os, pathlib, re, shutil, socket, subprocess, tempfile, time, unittest, urllib.error, urllib.parse, urllib.request, ssl, threading, email.parser, email.policy

PHP = os.environ.get("PHP_BIN", "php")
PASSWORD = "Test-owner-password-123!"
SETUP = "isolated-test-setup-token-32-characters-minimum"

class AppTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.temp = tempfile.TemporaryDirectory(prefix="deshi-test-")
        cls.root = pathlib.Path(cls.temp.name)
        source = pathlib.Path(__file__).resolve().parents[1]
        shutil.copytree(source / "public_html", cls.root / "public_html")
        shutil.copytree(source / "deshi_canvas_app", cls.root / "deshi_canvas_app", ignore=shutil.ignore_patterns(".env", "storage"))
        (cls.root / "deshi_canvas_app/storage/uploads").mkdir(parents=True)
        sock = socket.socket(); sock.bind(("127.0.0.1", 0)); cls.port = sock.getsockname()[1]; sock.close()
        cls.base = f"http://127.0.0.1:{cls.port}"
        cls.config = cls.root / "deshi_canvas_app/.env"
        mysql = os.environ.get("DESHI_TEST_MYSQL_DATABASE", "")
        if mysql and not mysql.startswith("deshi_test_"): raise RuntimeError("Use a disposable deshi_test_* database.")
        cls.config.write_text(f"APP_ENV=test\nAPP_URL={cls.base}\nDB_DRIVER={'mysql' if mysql else 'sqlite'}\nDB_DATABASE={mysql or str(cls.root / 'test.sqlite')}\nDB_HOST=127.0.0.1\nDB_PORT={os.environ.get('DESHI_TEST_MYSQL_PORT','3306')}\nDB_USERNAME={os.environ.get('DESHI_TEST_MYSQL_USER','root')}\nDB_PASSWORD={os.environ.get('DESHI_TEST_MYSQL_PASSWORD','')}\nSETUP_TOKEN={SETUP}\nWHATSAPP_APP_SECRET=isolated-test-meta-secret\nWHATSAPP_VERIFY_TOKEN=isolated-verify-token\n", encoding="utf-8")
        cls.log = open(cls.root / "server.log", "w")
        cls.server = subprocess.Popen([PHP, "-d", "upload_max_filesize=10M", "-d", "post_max_size=12M", "-S", f"127.0.0.1:{cls.port}", "-t", str(cls.root / "public_html")], stdout=cls.log, stderr=cls.log)
        cls.browser = cls.client()
        for _ in range(50):
            try:
                status, _, body = cls.req("/setup.php", client=cls.browser)
                if status == 200: break
            except (ConnectionError, urllib.error.URLError): pass
            time.sleep(.1)
        else: raise RuntimeError("PHP server did not start.")
        cls.req("/setup.php", {"csrf": cls.token(body), "setup_token": SETUP, "name": "Test Owner", "email": "owner@example.test", "password": PASSWORD, "confirm_password": PASSWORD}, cls.browser)
        result = cls.sql("SELECT COUNT(*) AS n FROM users WHERE role='owner'")
        if int(result[0]["n"]) != 1: raise RuntimeError("Owner installation failed: " + str(cls.root))

    @classmethod
    def tearDownClass(cls):
        cls.server.terminate(); cls.server.wait(timeout=5); cls.log.close()
        errors = (cls.root / "deshi_canvas_app/storage/php-errors.log")
        if errors.exists() and errors.read_text().strip(): print("PHP error log:", errors.read_text())
        cls.temp.cleanup()

    def setUp(self): self.sql("DELETE FROM rate_limits")

    @staticmethod
    def client(): return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))

    @classmethod
    def req(cls, path, data=None, client=None, raw=None, headers=None):
        headers = headers or {}
        body = raw if raw is not None else (urllib.parse.urlencode(data).encode() if data is not None else None)
        if data is not None: headers.setdefault("Content-Type", "application/x-www-form-urlencoded")
        request = urllib.request.Request(cls.base + path, data=body, headers=headers)
        try:
            with (client or cls.client()).open(request, timeout=30) as r: return r.status, r.headers, r.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as r: return r.code, r.headers, r.read().decode("utf-8", "replace")

    @staticmethod
    def token(body, name="csrf"):
        match = re.search(r'name="' + re.escape(name) + r'" value="([^"]*)"', body)
        if not match: raise AssertionError("Token not found: " + name + " " + body[:300])
        return match[1]

    @classmethod
    def php(cls, body, ini=()):
        init = json.dumps(str(cls.root / "deshi_canvas_app/bootstrap.php"))
        flags=[flag for value in ini for flag in ["-d",value]]
        result = subprocess.run([PHP, *flags, "-r", "require " + init + ";" + body], capture_output=True, text=True)
        if result.returncode: raise AssertionError(result.stderr + result.stdout)
        return result.stdout

    @classmethod
    def sql(cls, statement, params=()):
        sql = base64.b64encode(statement.encode()).decode()
        par = base64.b64encode(json.dumps(list(params)).encode()).decode()
        code = f"$s=query(base64_decode('{sql}'),json_decode(base64_decode('{par}'),true)); echo json_encode($s->columnCount()?$s->fetchAll():[]);"
        return json.loads(cls.php(code))

    def login(self, email="owner@example.test", password=PASSWORD):
        browser = self.client(); _, _, html = self.req("/admin/login.php", client=browser)
        status, _, html = self.req("/admin/login.php", {"csrf":self.token(html), "email":email, "password":password}, browser)
        self.assertEqual(status, 200); return browser, html

    def order(self, wa=True, extra=None):
        browser = self.client(); status, _, html = self.req("/", client=browser); self.assertEqual(status, 200)
        data = {"csrf":self.token(html), "submission_token":self.token(html,"submission_token"), "name":"Example Client", "email":"client@example.test", "phone":"+447123456789", "company":"Example Studio", "service_id":"1", "brief":"Please prepare and organise our spreadsheet.", "privacy_consent":"1"}
        if wa: data["wa_consent"]="1"
        data.update(extra or {})
        status, _, result = self.req("/api/order.php", data, browser)
        self.assertEqual(status, 200, result)
        parsed = json.loads(result)
        return browser, data, parsed

    def test_public_site_and_installer_lock(self):
        status, headers, html = self.req("/")
        self.assertEqual(status, 200)
        for name in ["Deshi Canvas","Data Entry","Graphic Design","Video Editing","Motion Graphics","Digital Marketing","Social Media Marketing","+8801751247714"]: self.assertIn(name, html)
        self.assertNotIn("Brandora", html); self.assertNotIn("$1,775", html)
        self.assertIn("script-src 'self'", headers["Content-Security-Policy"])
        self.assertEqual(self.req("/assets/site.css")[0],200)
        self.assertEqual(self.req("/setup.php")[0],403)
        self.assertEqual(self.req("/.env")[0],404)
        self.assertEqual(self.req("/deshi_canvas_app/.env")[0],404)

    def test_catalog_editing_preserves_order_history(self):
        owner,html=self.login()
        values={"csrf":self.token(html),"action":"save_service","name":"Catalog QA Service","description":"Research and catalogue organisation for a test project.","tags":"Research,Planning","image":"images/service-product.webp","sort_order":"9","active":"1"}
        self.assertEqual(self.req("/admin/?page=services",values,owner)[0],200)
        service=self.sql("SELECT id FROM services WHERE name=?",["Catalog QA Service"])[0]["id"]
        self.assertIn("Catalog QA Service",self.req("/")[2])
        _,_,made=self.order(extra={"service_id":str(service)})
        values.update({"record_id":str(service),"name":"Hidden QA Service"});values.pop("active")
        self.req("/admin/?page=services",values,owner)
        self.assertNotIn("Hidden QA Service",self.req("/")[2])
        self.assertEqual(self.sql("SELECT service_name FROM orders WHERE reference=?",[made["reference"]])[0]["service_name"],"Catalog QA Service")
        project={"csrf":self.token(html),"action":"save_project","title":"QA Portfolio Item","category":"Visual concept","image":"images/portfolio-1.webp","sort_order":"9","active":"1"}
        self.req("/admin/?page=portfolio",project,owner)
        self.assertIn("QA Portfolio Item",self.req("/")[2])

    def test_order_idempotency_and_opt_out(self):
        before=int(self.sql("SELECT COUNT(*) AS n FROM orders")[0]["n"])
        browser, data, first=self.order(wa=False)
        second=json.loads(self.req("/api/order.php",data,browser)[2])
        self.assertEqual(first["reference"],second["reference"])
        self.assertEqual(int(self.sql("SELECT COUNT(*) AS n FROM orders")[0]["n"]),before+1)
        self.assertEqual(self.sql("SELECT status FROM notifications WHERE order_id=(SELECT id FROM orders WHERE reference=?) AND channel='whatsapp'",[first["reference"]])[0]["status"],"skipped")

    def test_csrf_validation_and_inactive_service(self):
        browser,data,_=self.order()
        bad=dict(data,csrf="bad"); self.assertEqual(self.req("/api/order.php",bad,browser)[0],419)
        _,_,html=self.req("/",client=browser)
        bad=dict(data,submission_token=self.token(html,"submission_token"),phone="01751247714")
        self.assertEqual(self.req("/api/order.php",bad,browser)[0],422)
        bad.update(phone="+447123456789", service_id="999999")
        self.assertEqual(self.req("/api/order.php",bad,browser)[0],422)
        self.assertEqual(self.req("/api/order.php")[0],405)

    def test_upload_and_private_download(self):
        browser=self.client(); _,_,html=self.req("/",client=browser)
        fields={"csrf":self.token(html),"submission_token":self.token(html,"submission_token"),"name":"Attachment Test","email":"file@example.test","phone":"+447123456789","service_id":"1","brief":"Please review the attached project notes.","privacy_consent":"1"}
        boundary="DeshiTestBoundary124"
        def upload(filename,mime,content):
            bits=[]
            for k,v in fields.items(): bits.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n')
            bits.append(f'--{boundary}\r\nContent-Disposition: form-data; name="attachment"; filename="{filename}"\r\nContent-Type: {mime}\r\n\r\n{content}\r\n--{boundary}--\r\n')
            return self.req("/api/order.php",client=browser,raw="".join(bits).encode(),headers={"Content-Type":"multipart/form-data; boundary="+boundary})
        self.assertEqual(upload("bad.php","text/plain","<?php echo 1;")[0],422)
        status,_,body=upload("brief.txt","text/plain","private project notes")
        self.assertEqual(status,200,body)
        ref=json.loads(body)["reference"]; order=self.sql("SELECT * FROM orders WHERE reference=?",[ref])[0]
        self.assertTrue(order["attachment_path"].endswith(".bin"))
        self.assertEqual(self.req("/admin/download.php?id="+str(order["id"]))[0],200) # redirected to sign-in
        owner,_=self.login(); status,headers,content=self.req("/admin/download.php?id="+str(order["id"]),client=owner)
        self.assertEqual(status,200); self.assertEqual(content,"private project notes")
        self.assertIn("attachment",headers["Content-Disposition"])
        self.assertEqual(self.req("/uploads/"+order["attachment_path"])[0],404)

    def test_editor_access_and_assignment(self):
        owner,html=self.login()
        self.req("/admin/?page=staff",{"csrf":self.token(html),"action":"create_staff","name":"Test Editor","email":"editor@example.test","role":"editor","temporary_password":"Temporary-editor-pass-123!","owner_password":PASSWORD},owner)
        editorid=int(self.sql("SELECT id FROM users WHERE email=?",["editor@example.test"])[0]["id"])
        editor,html=self.login("editor@example.test","Temporary-editor-pass-123!")
        self.assertIn("Choose your own password",html)
        self.req("/admin/?page=account",{"csrf":self.token(html),"action":"change_password","current_password":"Temporary-editor-pass-123!","new_password":"Changed-editor-password-123!","confirm_password":"Changed-editor-password-123!"},editor)
        _,_,first=self.order(extra={"name":"Assigned Client"}); _,_,second=self.order(extra={"name":"Hidden Client"})
        oid=int(self.sql("SELECT id FROM orders WHERE reference=?",[first["reference"]])[0]["id"]); other=int(self.sql("SELECT id FROM orders WHERE reference=?",[second["reference"]])[0]["id"])
        _,_,html=self.req("/admin/?page=orders&id="+str(oid),client=owner)
        self.req("/admin/?page=orders&id="+str(oid),{"csrf":self.token(html),"action":"update_order","order_id":oid,"status":"contacted","assigned_to":editorid},owner)
        status,_,html=self.req("/admin/?page=orders",client=editor)
        self.assertIn("Assigned Client",html); self.assertNotIn("Hidden Client",html)
        self.assertEqual(self.req("/admin/?page=orders&id="+str(other),client=editor)[0],404)
        self.assertEqual(self.req("/admin/?page=settings",client=editor)[0],403)
        _,_,html=self.req("/admin/?page=orders&id="+str(oid),client=editor)
        self.req("/admin/?page=orders&id="+str(oid),{"csrf":self.token(html),"action":"update_order","order_id":oid,"status":"in_progress","assigned_to":1},editor)
        saved=self.sql("SELECT status,assigned_to FROM orders WHERE id=?",[oid])[0]
        self.assertEqual(saved["status"],"in_progress"); self.assertEqual(int(saved["assigned_to"]),editorid)
        self.assertEqual(self.req("/admin/?page=orders",{"csrf":self.token(html),"action":"create_staff"},editor)[0],403)
        self.sql("UPDATE users SET active=0,auth_version=auth_version+1 WHERE id=?",[editorid])
        self.assertIn("Welcome back.",self.req("/admin/",client=editor)[2])

    def test_notifications_durable_and_retryable(self):
        _,_,made=self.order(); oid=int(self.sql("SELECT id FROM orders WHERE reference=?",[made["reference"]])[0]["id"])
        self.php(f"process_notifications(30,{oid});")
        blocked=self.sql("SELECT status,attempts FROM notifications WHERE order_id=?",[oid])
        self.assertTrue(all(x["status"]=="blocked" and int(x["attempts"])==0 for x in blocked))
        self.php(f"retry_notifications({oid});$GLOBALS['notification_transport']=function($c,$o){{throw new RuntimeException('private-token-not-to-be-logged');}};process_notifications(30,{oid});")
        failed=self.sql("SELECT status,error_code FROM notifications WHERE order_id=?",[oid])
        self.assertTrue(all(x["status"]=="retry" for x in failed)); self.assertNotIn("private-token",str(failed))
        self.php(f"retry_notifications({oid});$GLOBALS['notification_transport']=fn($c,$o)=>'mock-'.$c.'-'.$o['id'];process_notifications(30,{oid});process_notifications(30,{oid});")
        saved=self.sql("SELECT status,attempts FROM notifications WHERE order_id=?",[oid])
        self.assertTrue(all(x["status"]=="accepted" and int(x["attempts"])==1 for x in saved))
        self.php(f"retry_notifications({oid});")
        self.assertTrue(all(x["status"]=="accepted" for x in self.sql("SELECT status FROM notifications WHERE order_id=?",[oid])))

    def test_signed_webhook_and_ordered_delivery(self):
        _,_,made=self.order(); oid=int(self.sql("SELECT id FROM orders WHERE reference=?",[made["reference"]])[0]["id"])
        self.php(f"$GLOBALS['notification_transport']=fn($c,$o)=>'webhook-'.$c.'-'.$o['id'];process_notifications(30,{oid});")
        def event(state,valid=True):
            body=json.dumps({"entry":[{"changes":[{"value":{"statuses":[{"id":"webhook-whatsapp-"+str(oid),"status":state}]}}]}]}).encode()
            signature="sha256="+hmac.new(b"isolated-test-meta-secret",body,hashlib.sha256).hexdigest() if valid else "sha256=invalid"
            return self.req("/api/whatsapp-webhook.php",raw=body,headers={"Content-Type":"application/json","X-Hub-Signature-256":signature})[0]
        self.assertEqual(event("delivered",False),403); self.assertEqual(event("delivered"),200); self.assertEqual(event("sent"),200)
        self.assertEqual(self.sql("SELECT status FROM notifications WHERE order_id=? AND channel='whatsapp'",[oid])[0]["status"],"delivered")
        self.assertEqual(self.req("/api/whatsapp-webhook.php?hub.mode=subscribe&hub.verify_token=isolated-verify-token&hub.challenge=12345")[2],"12345")

    def test_settings_social_links_and_escaping(self):
        owner,html=self.login()
        values={r["name"]:r["value"] for r in self.sql("SELECT name,value FROM settings")}
        values.update({"csrf":self.token(html),"action":"save_settings","social_linkedin":"https://www.linkedin.com/company/example-canvas/","social_instagram":"https://www.instagram.com/example-canvas/"})
        self.req("/admin/?page=settings",values,owner)
        self.assertIn(values["social_linkedin"],self.req("/")[2])
        output=self.php("echo confirmation_content(['name'=>'<script>test</script>','reference'=>'DC-TEST','service_name'=>'Graphic Design'])['html'];")
        self.assertIn("LinkedIn",output); self.assertIn("&lt;script&gt;",output); self.assertNotIn("<script>",output)
        values["social_facebook"]="javascript:alert(1)"
        _,_,html=self.req("/admin/?page=settings",client=owner); values["csrf"]=self.token(html)
        self.req("/admin/?page=settings",values,owner)
        self.assertEqual(self.sql("SELECT value FROM settings WHERE name='social_linkedin'")[0]["value"],"https://www.linkedin.com/company/example-canvas/")
        self.assertNotIn("javascript:alert",self.req("/")[2])

    def test_whatsapp_stop_reply_cancels_pending_messages(self):
        phone="+447123450001"
        _,_,made=self.order(extra={"phone":phone})
        body=json.dumps({"entry":[{"changes":[{"value":{"messages":[{"from":phone[1:],"text":{"body":"STOP"}}]}}]}]}).encode()
        signature="sha256="+hmac.new(b"isolated-test-meta-secret",body,hashlib.sha256).hexdigest()
        self.assertEqual(self.req("/api/whatsapp-webhook.php",raw=body,headers={"Content-Type":"application/json","X-Hub-Signature-256":signature})[0],200)
        self.assertEqual(int(self.sql("SELECT wa_consent FROM orders WHERE reference=?",[made["reference"]])[0]["wa_consent"]),0)
        self.assertEqual(self.sql("SELECT status FROM notifications WHERE order_id=(SELECT id FROM orders WHERE reference=?) AND channel='whatsapp'",[made["reference"]])[0]["status"],"skipped")

    def test_ssl_smtp_confirmation_transport(self):
        cert=self.root/"smtp-cert.pem"; key=self.root/"smtp-key.pem"
        subprocess.run(["openssl","req","-x509","-newkey","rsa:2048","-nodes","-days","1","-subj","/CN=127.0.0.1","-addext","subjectAltName=IP:127.0.0.1","-keyout",str(key),"-out",str(cert)],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
        listener=socket.socket();listener.bind(("127.0.0.1",0));listener.listen(1);listener.settimeout(15)
        port=listener.getsockname()[1]; captured=[]; errors=[]
        def smtp_server():
            try:
                context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER);context.load_cert_chain(cert,key)
                conn,_=listener.accept()
                with context.wrap_socket(conn,server_side=True) as sock:
                    stream=sock.makefile("rwb")
                    def send(line):stream.write(line+b"\r\n");stream.flush()
                    send(b"220 localhost ESMTP test")
                    auth=0
                    while True:
                        line=stream.readline()
                        if not line:break
                        command=line.upper().strip()
                        if auth==1:send(b"334 UGFzc3dvcmQ6");auth=2
                        elif auth==2:send(b"235 Authentication successful");auth=0
                        elif command.startswith(b"EHLO"):send(b"250-localhost\r\n250-AUTH LOGIN PLAIN\r\n250 SIZE 2000000")
                        elif command.startswith(b"AUTH LOGIN"):send(b"334 VXNlcm5hbWU6");auth=1
                        elif command.startswith(b"AUTH PLAIN"):send(b"235 Authentication successful")
                        elif command.startswith((b"MAIL FROM",b"RCPT TO",b"RSET")):send(b"250 OK")
                        elif command==b"DATA":
                            send(b"354 Send message")
                            parts=[]
                            while True:
                                piece=stream.readline()
                                if piece==b".\r\n" or not piece:break
                                parts.append(piece)
                            captured.append(b"".join(parts));send(b"250 Queued")
                        elif command==b"QUIT":send(b"221 Bye");break
                        else:send(b"250 OK")
            except Exception as exc:errors.append(str(exc))
            finally:listener.close()
        thread=threading.Thread(target=smtp_server,daemon=True);thread.start()
        config={"MAIL_HOST":"127.0.0.1","MAIL_PORT":str(port),"MAIL_ENCRYPTION":"ssl","MAIL_USERNAME":"orders@example.test","MAIL_PASSWORD":"fake-test-smtp-password","MAIL_FROM":"orders@example.test"}
        encoded=base64.b64encode(json.dumps(config).encode()).decode()
        code="ini_set('openssl.cafile',"+json.dumps(str(cert))+");$GLOBALS['env']=array_merge($GLOBALS['env'],json_decode(base64_decode('"+encoded+"'),true));echo deliver_notification('email',['name'=>'SMTP Test Client','email'=>'client@example.test','reference'=>'DC-SMTP-TEST','service_name'=>'Graphic Design']);"
        self.assertTrue(self.php(code,ini=["openssl.cafile="+str(cert)]));thread.join(5)
        self.assertEqual(errors,[]);self.assertEqual(len(captured),1)
        message=email.parser.BytesParser(policy=email.policy.default).parsebytes(captured[0])
        content="\n".join(part.get_content() for part in message.walk() if part.get_content_type() in ["text/plain","text/html"])
        self.assertIn("info@deshicanvas.com",content);self.assertIn("+8801751247714",content);self.assertIn("DC-SMTP-TEST",content)

    def test_xss_sql_search_csrf_and_logout(self):
        _,_,made=self.order(extra={"name":"<svg onload=alert(1)>Test"})
        oid=int(self.sql("SELECT id FROM orders WHERE reference=?",[made["reference"]])[0]["id"])
        owner,html=self.login(); detail=self.req("/admin/?page=orders&id="+str(oid),client=owner)[2]
        self.assertIn("&lt;svg",detail); self.assertNotIn("<svg onload",detail)
        self.assertEqual(self.req("/admin/?page=orders&q="+urllib.parse.quote("' OR 1=1 --"),client=owner)[0],200)
        self.assertEqual(self.req("/admin/?page=orders",{"csrf":"bad","action":"add_note","order_id":oid,"note":"bad"},owner)[0],419)
        self.assertEqual(self.req("/admin/logout.php",client=owner)[0],405)
        self.assertIn("Welcome back.",self.req("/admin/logout.php",{"csrf":self.token(html)},owner)[2])

    def test_password_change_revokes_other_sessions(self):
        email="session@example.test"; pw="Session-test-password-123!"
        self.php("$h=password_hash('"+pw+"',PASSWORD_DEFAULT);query('INSERT INTO users (name,email,password_hash,role,created_at) VALUES (?,?,?,?,?)',['Session Test','"+email+"',$h,'admin',now()]);")
        a,html=self.login(email,pw); b,_=self.login(email,pw)
        self.req("/admin/?page=account",{"csrf":self.token(html),"action":"change_password","current_password":pw,"new_password":"New-session-password-123!","confirm_password":"New-session-password-123!"},a)
        self.assertIn("Welcome back.",self.req("/admin/",client=b)[2])
        self.assertIn("Your workspace.",self.req("/admin/",client=a)[2])

    def test_canonical_production_https_requirement(self):
        content=self.config.read_text()
        try:
            self.config.write_text(content.replace("APP_ENV=test","APP_ENV=production"))
            self.assertEqual(self.req("/admin/login.php")[0],400)
        finally: self.config.write_text(content)

if __name__=="__main__": unittest.main(verbosity=2)