MarketingTool/apps/api/tests/test_provider_config.py

import json, os, sqlite3, stat, threading, unittest
from http.client import HTTPConnection
from tempfile import TemporaryDirectory
from unittest.mock import patch
from app.main import create_server, hash_password

class ProviderConfigTests(unittest.TestCase):
    def setUp(self):
        self.tmp=TemporaryDirectory(); self.old_env={k:os.environ.get(k) for k in ('BOOTSTRAP_ADMIN_EMAIL','BOOTSTRAP_ADMIN_PASSWORD','PROVIDER_CONFIG_KEY_FILE','AI_RESEARCH_PROVIDER')}; os.environ['BOOTSTRAP_ADMIN_EMAIL']='pc-owner@test'; os.environ['BOOTSTRAP_ADMIN_PASSWORD']='pw'; os.environ['PROVIDER_CONFIG_KEY_FILE']=self.tmp.name+'/provider.key'
        self.db=self.tmp.name+'/db.sqlite'; self.server=create_server('127.0.0.1',0,self.db); self.thread=threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start(); self.conn=HTTPConnection('127.0.0.1',self.server.server_port); self.cookie=None; self.req('POST','/api/v1/auth/login',{'email':'pc-owner@test','password':'pw'})
    def tearDown(self):
        self.server.shutdown(); self.server.server_close(); self.thread.join(2)
        for k,v in self.old_env.items():
            if v is None: os.environ.pop(k,None)
            else: os.environ[k]=v
        self.tmp.cleanup()
    def req(self,m,p,b=None):
        body=json.dumps(b).encode() if b is not None else None; h={'Content-Type':'application/json'}
        if self.cookie:h['Cookie']=self.cookie
        self.conn.request(m,p,body,h); r=self.conn.getresponse(); c=r.getheader('Set-Cookie');
        if c:self.cookie=c.split(';',1)[0]
        return r.status,json.loads(r.read() or b'{}')
    def test_admin_can_store_write_only_encrypted_config_and_read_status(self):
        payload={'provider':'nous_portal','model':'Hermes-test','enabled':True,'credentials':{'nous_api_key':'nous-secret','firecrawl_api_key':'fire-secret'}}
        status,out=self.req('POST','/api/v1/admin/ai-provider-config',payload); self.assertEqual(status,200); self.assertNotIn('secret',json.dumps(out)); self.assertEqual(out['source'],'database'); self.assertTrue(out['configured'])
        with open(self.db,'rb') as handle: raw=handle.read()
        self.assertNotIn(b'nous-secret',raw); self.assertNotIn(b'fire-secret',raw)
        mode=stat.S_IMODE(os.stat(os.environ['PROVIDER_CONFIG_KEY_FILE']).st_mode); self.assertEqual(mode,0o600)
        status,out=self.req('GET','/api/v1/admin/ai-provider-config'); self.assertNotIn('credentials',out); self.assertEqual(out['status'],'ready')
    def test_patch_ui_contract_accepts_nous_and_internal_searxng_without_firecrawl(self):
        status,out=self.req('PATCH','/api/v1/ai/provider-config',{'provider':'nous_portal','model':'Hermes-4-405B','nous_base_url':'https://inference-api.nousresearch.com/v1','searxng_base_url':'http://searxng:8080','enabled':True,'credentials':{'nous_api_key':'nous-only'}})
        self.assertEqual(status,200); self.assertTrue(out['configured']); self.assertEqual(out['provider'],'nous_portal')

    def test_viewer_cannot_mutate_but_can_read_safe_status(self):
        ph,s=hash_password('viewer'); db=sqlite3.connect(self.db); db.execute("INSERT INTO users(organization_id,email,password_hash,password_salt,role) VALUES(?,?,?,?,?)",('demo-tenant','pc-viewer@test',ph,s,'viewer')); db.commit(); db.close(); self.cookie=None; self.req('POST','/api/v1/auth/login',{'email':'pc-viewer@test','password':'viewer'})
        self.assertEqual(self.req('GET','/api/v1/admin/ai-provider-config')[0],200); self.assertEqual(self.req('POST','/api/v1/admin/ai-provider-config',{'provider':'nous_portal'})[0],403)
    def test_invalid_config_and_connectivity_never_sends_outreach(self):
        self.assertEqual(self.req('POST','/api/v1/admin/ai-provider-config',{'provider':'nous_portal','enabled':True,'credentials':{'firecrawl_api_key':'y'}})[0],400)
        self.req('POST','/api/v1/admin/ai-provider-config',{'provider':'nous_portal','enabled':True,'credentials':{'nous_api_key':'x','firecrawl_api_key':'y'}})
        with patch('app.provider_config.urlopen') as opened:
            status,out=self.req('POST','/api/v1/admin/ai-provider-config/test',{}); self.assertEqual(status,200); self.assertFalse(out['outbound_calls']); self.assertEqual(out['network_calls'],2); self.assertEqual(opened.call_count,2)
            for call in opened.call_args_list: self.assertEqual(call.args[0].method,'GET')
    def test_persists_after_server_restart_and_db_config_beats_env(self):
        self.req('POST','/api/v1/admin/ai-provider-config',{'provider':'nous_portal','enabled':True,'credentials':{'nous_api_key':'x','firecrawl_api_key':'y'}}); self.server.shutdown(); self.server.server_close(); self.thread.join(2)
        os.environ['AI_RESEARCH_PROVIDER']='untrusted'; self.server=create_server('127.0.0.1',0,self.db); self.thread=threading.Thread(target=self.server.serve_forever,daemon=True); self.thread.start(); self.conn=HTTPConnection('127.0.0.1',self.server.server_port); self.cookie=None; self.req('POST','/api/v1/auth/login',{'email':'pc-owner@test','password':'pw'}); status,out=self.req('GET','/api/v1/admin/ai-provider-config'); self.assertEqual(status,200); self.assertEqual(out['provider'],'nous_portal')

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