#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2009 Fermi Research Alliance, LLC
# SPDX-License-Identifier: Apache-2.0
"""
Project:
glideinWMS
Description:
unit tests for glideinwms/frontend/glideinFrontendPlugins.py
Author:
Doug Strain
"""
import os
# import sys
# import shutil
import tempfile
# import tarfile
import unittest
import xmlrunner
from glideinwms.lib import condorMonitor, logSupport
# from glideinwms.unittests.unittest_utils import create_temp_file
# from glideinwms.unittests.unittest_utils import create_random_string
from glideinwms.unittests.unittest_utils import FakeLogger, TestImportError
try:
from glideinwms.frontend import glideinFrontendPlugins
from glideinwms.frontend.glideinFrontendInterface import Credential
except ImportError as err:
raise TestImportError(str(err))
[docs]
class fakeObj:
def __init__(self):
self.min_nr_glideins = 10
self.max_run_glideins = 30
[docs]
class fakeDescript:
def __init__(self):
self.merged_data = {
"ProxySecurityClasses": {},
"ProxyTrustDomains": {},
"ProxyTypes": {},
"CredentialGenerators": {},
"ProxyKeyFiles": {},
"ProxyPilotFiles": {},
"ProxyVMIds": {},
"ProxyVMTypes": {},
"ProxyVMIdFname": {},
"ProxyVMTypeFname": {},
"ProxyRemoteUsernames": {},
"ProxyProjectIds": {},
"ProxyCreationScripts": {},
"ProxyUpdateFrequency": {},
}
[docs]
def addproxy(self, name):
self.merged_data["ProxySecurityClasses"][name] = "frontend"
self.merged_data["ProxyTrustDomains"][name] = "OSG"
self.merged_data["ProxyTypes"][name] = "grid_proxy"
# self.merged_data['ProxyKeyFiles'][name]=
# self.merged_data['ProxyPilotFiles'][name]=""
# self.merged_data['ProxyVMIds'][name]=""
# self.merged_data['ProxyVMTypes'][name]=""
[docs]
class TestPlugins(unittest.TestCase):
"""
Test the proxy plugins
"""
[docs]
def getCredlist(self):
# Create fake Credentials
rtnlist = []
for t in range(30):
(f, proxyfile) = tempfile.mkstemp()
os.close(f)
self.elementDescript.addproxy(proxyfile)
rtnlist.append(Credential(t, proxyfile, self.elementDescript))
return rtnlist
[docs]
def killCredlist(self, cred_list):
for cred in cred_list:
os.remove(cred.filename)
[docs]
def setUp(self):
logSupport.log = FakeLogger()
self.config_dir = "/tmp"
# self.working_dir = tempfile.mkdtemp()
self.credlist = []
# Create fake descript
self.elementDescript = fakeDescript()
self.createCondor()
# Fake calling condor_q and condor_status
[docs]
def createCondor(self):
self.condorq_dict = {}
self.condorstatus_dict = {}
key = "schedd_job1@schedd1.domain.tld"
key2 = "schedd1.domain.tld"
key3 = "usercollector_service@schedd1.domain.tld"
self.condorq_dict[key] = condorMonitor.CondorQ(key, schedd_lookup_cache=None)
self.condorq_dict[key].stored_data = {}
for jid in range(10):
user_str = "user" + str(jid) + "@random.domain.tld"
self.condorq_dict[key].stored_data[(jid, 0)] = {}
self.condorq_dict[key].stored_data[(jid, 0)]["User"] = user_str
self.condorq_dict[key].stored_data[(jid, 0)]["ClusterId"] = 0
self.condorq_dict[key].stored_data[(jid, 0)]["ProcId"] = jid
self.condorq_dict[key].stored_data[(jid, 0)]["JobStatus"] = 1
self.condorstatus_dict[user_str] = {}
self.condorstatus_dict[user_str]["MyType"] = "Submitter"
entry_str = "entry_name@instance@service@vofrontend_service-vofrontend_instance.main"
self.condorstatus_dict[entry_str] = {}
self.condorstatus_dict[entry_str]["MyType"] = "glideresource"
self.condorstatus_dict[key] = {}
self.condorstatus_dict[key]["MyType"] = "Scheduler"
self.condorstatus_dict[key]["NumUsers"] = 0
self.condorstatus_dict[key2] = {}
self.condorstatus_dict[key2]["MyType"] = "Negotiator"
self.condorstatus_dict[key3] = {}
self.condorstatus_dict[key3]["MyType"] = "Collector"
for k in list(self.condorstatus_dict.keys()):
self.condorstatus_dict[k]["CurrentTime"] = "time()"
[docs]
def tearDown(self):
if os.path.exists("/tmp/proxy_usermap_wr.dat"):
os.remove("/tmp/proxy_usermap_wr.dat")
if os.path.exists("/tmp/proxy_user_rr.dat"):
os.remove("/tmp/proxy_user_rr.dat")
[docs]
def test_proxy_first(self):
self.credlist = self.getCredlist()
p = glideinFrontendPlugins.ProxyFirst(self.config_dir, self.credlist)
p.update_usermap(self.condorq_dict, self.condorq_dict, self.condorstatus_dict, self.condorstatus_dict)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "OSG")
self.assertEqual(testlist, [self.credlist[0]])
testlist = p.get_credentials(fakeObj(), "grid_proxy", "FAKE")
self.assertEqual(testlist, [])
self.killCredlist(self.credlist)
[docs]
def test_proxy_all(self):
self.credlist = self.getCredlist()
p = glideinFrontendPlugins.ProxyAll(self.config_dir, self.credlist)
p.update_usermap(self.condorq_dict, self.condorq_dict, self.condorstatus_dict, self.condorstatus_dict)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "OSG")
self.assertEqual(testlist, self.credlist)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "FAKE")
self.assertEqual(testlist, [])
self.killCredlist(self.credlist)
[docs]
def test_proxy_userr(self):
self.credlist = self.getCredlist()
p = glideinFrontendPlugins.ProxyUserRR(self.config_dir, self.credlist)
p.update_usermap(self.condorq_dict, self.condorq_dict, self.condorstatus_dict, self.condorstatus_dict)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "OSG")
self.assertEqual(testlist, self.credlist[0:10])
self.killCredlist(self.credlist)
[docs]
def test_proxy_cardinality(self):
self.credlist = self.getCredlist()
p = glideinFrontendPlugins.ProxyUserCardinality(self.config_dir, self.credlist)
p.update_usermap(self.condorq_dict, self.condorq_dict, self.condorstatus_dict, self.condorstatus_dict)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "OSG")
self.assertEqual(testlist, self.credlist[0:10])
self.killCredlist(self.credlist)
[docs]
def test_proxy_usermap(self):
self.credlist = self.getCredlist()
p = glideinFrontendPlugins.ProxyUserMapWRecycling(self.config_dir, self.credlist)
p.update_usermap(self.condorq_dict, self.condorq_dict, self.condorstatus_dict, self.condorstatus_dict)
testlist = p.get_credentials(fakeObj(), "grid_proxy", "OSG")
self.assertEqual(testlist, self.credlist[0:10])
self.killCredlist(self.credlist)
if __name__ == "__main__":
unittest.main(testRunner=xmlrunner.XMLTestRunner(output="unittests-reports"))