summaryrefslogtreecommitdiff
path: root/BaseTools/Source/Python/UPT/Core
diff options
context:
space:
mode:
authorkx <kx@radix.pro>2024-06-04 06:43:08 +0300
committerkx <kx@radix.pro>2024-06-04 06:43:08 +0300
commit7e2ccccace24636f29ddc210b94606abd4c7e42b (patch)
tree545d43ea12671048956cafeb12303e84d601e571 /BaseTools/Source/Python/UPT/Core
parent27b044605cd5f6b33a3d231576003850b3fe305b (diff)
downloadedk2-trunk.tar.xz
Renormalized end-of-lines from master@27b044605cd5f6b33a3d231576003850b3fe305btrunk
Diffstat (limited to 'BaseTools/Source/Python/UPT/Core')
-rw-r--r--BaseTools/Source/Python/UPT/Core/DependencyRules.py896
-rw-r--r--BaseTools/Source/Python/UPT/Core/DistributionPackageClass.py534
-rw-r--r--BaseTools/Source/Python/UPT/Core/FileHook.py386
-rw-r--r--BaseTools/Source/Python/UPT/Core/IpiDb.py1844
-rw-r--r--BaseTools/Source/Python/UPT/Core/PackageFile.py500
-rw-r--r--BaseTools/Source/Python/UPT/Core/__init__.py28
6 files changed, 2094 insertions, 2094 deletions
diff --git a/BaseTools/Source/Python/UPT/Core/DependencyRules.py b/BaseTools/Source/Python/UPT/Core/DependencyRules.py
index 0c801c72d6..2742db847a 100644
--- a/BaseTools/Source/Python/UPT/Core/DependencyRules.py
+++ b/BaseTools/Source/Python/UPT/Core/DependencyRules.py
@@ -1,448 +1,448 @@
-## @file
-# This file is for installed package information database operations
-#
-# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-#
-
-'''
-Dependency
-'''
-
-##
-# Import Modules
-#
-from os.path import dirname
-import os
-
-import Logger.Log as Logger
-from Logger import StringTable as ST
-from Library.Parsing import GetWorkspacePackage
-from Library.Parsing import GetWorkspaceModule
-from Library.Parsing import GetPkgInfoFromDec
-from Library.Misc import GetRelativePath
-from Library import GlobalData
-from Logger.ToolError import FatalError
-from Logger.ToolError import EDK1_INF_ERROR
-from Logger.ToolError import UNKNOWN_ERROR
-(DEPEX_CHECK_SUCCESS, DEPEX_CHECK_MODULE_NOT_FOUND, \
-DEPEX_CHECK_PACKAGE_NOT_FOUND, DEPEX_CHECK_DP_NOT_FOUND) = (0, 1, 2, 3)
-
-
-## DependencyRules
-#
-# This class represents the dependency rule check mechanism
-#
-# @param object: Inherited from object class
-#
-class DependencyRules(object):
- def __init__(self, Datab, ToBeInstalledPkgList=None):
- self.IpiDb = Datab
- self.WsPkgList = GetWorkspacePackage()
- self.WsModuleList = GetWorkspaceModule()
-
- self.PkgsToBeDepend = [(PkgInfo[1], PkgInfo[2]) for PkgInfo in self.WsPkgList]
-
- # Add package info from the DIST to be installed.
- self.PkgsToBeDepend.extend(self.GenToBeInstalledPkgList(ToBeInstalledPkgList))
-
- def GenToBeInstalledPkgList(self, ToBeInstalledPkgList):
- if not ToBeInstalledPkgList:
- return []
- RtnList = []
- for Dist in ToBeInstalledPkgList:
- for Package in Dist.PackageSurfaceArea:
- RtnList.append((Package[0], Package[1]))
-
- return RtnList
-
- ## Check whether a module exists by checking the Guid+Version+Name+Path combination
- #
- # @param Guid: Guid of a module
- # @param Version: Version of a module
- # @param Name: Name of a module
- # @param Path: Path of a module
- # @return: True if module existed, else False
- #
- def CheckModuleExists(self, Guid, Version, Name, Path):
- Logger.Verbose(ST.MSG_CHECK_MODULE_EXIST)
- ModuleList = self.IpiDb.GetModInPackage(Guid, Version, Name, Path)
- ModuleList.extend(self.IpiDb.GetStandaloneModule(Guid, Version, Name, Path))
- Logger.Verbose(ST.MSG_CHECK_MODULE_EXIST_FINISH)
- if len(ModuleList) > 0:
- return True
- else:
- return False
-
- ## Check whether a module depex satisfied.
- #
- # @param ModuleObj: A module object
- # @param DpObj: A distribution object
- # @return: True if module depex satisfied
- # False else
- #
- def CheckModuleDepexSatisfied(self, ModuleObj, DpObj=None):
- Logger.Verbose(ST.MSG_CHECK_MODULE_DEPEX_START)
- Result = True
- Dep = None
- if ModuleObj.GetPackageDependencyList():
- Dep = ModuleObj.GetPackageDependencyList()[0]
- for Dep in ModuleObj.GetPackageDependencyList():
- #
- # first check whether the dependency satisfied by current workspace
- #
- Exist = self.CheckPackageExists(Dep.GetGuid(), Dep.GetVersion())
- #
- # check whether satisfied by current distribution
- #
- if not Exist:
- if DpObj is None:
- Result = False
- break
- for GuidVerPair in DpObj.PackageSurfaceArea.keys():
- if Dep.GetGuid() == GuidVerPair[0]:
- if Dep.GetVersion() is None or \
- len(Dep.GetVersion()) == 0:
- Result = True
- break
- if Dep.GetVersion() == GuidVerPair[1]:
- Result = True
- break
- else:
- Result = False
- break
-
- if not Result:
- Logger.Error("CheckModuleDepex", UNKNOWN_ERROR, \
- ST.ERR_DEPENDENCY_NOT_MATCH % (ModuleObj.GetName(), \
- Dep.GetPackageFilePath(), \
- Dep.GetGuid(), \
- Dep.GetVersion()))
- return Result
-
- ## Check whether a package exists in a package list specified by PkgsToBeDepend.
- #
- # @param Guid: Guid of a package
- # @param Version: Version of a package
- # @return: True if package exist
- # False else
- #
- def CheckPackageExists(self, Guid, Version):
- Logger.Verbose(ST.MSG_CHECK_PACKAGE_START)
- Found = False
- for (PkgGuid, PkgVer) in self.PkgsToBeDepend:
- if (PkgGuid == Guid):
- #
- # if version is not empty and not equal, then not match
- #
- if Version and (PkgVer != Version):
- Found = False
- break
- else:
- Found = True
- break
- else:
- Found = False
-
- Logger.Verbose(ST.MSG_CHECK_PACKAGE_FINISH)
- return Found
-
- ## Check whether a package depex satisfied.
- #
- # @param PkgObj: A package object
- # @param DpObj: A distribution object
- # @return: True if package depex satisfied
- # False else
- #
- def CheckPackageDepexSatisfied(self, PkgObj, DpObj=None):
- ModuleDict = PkgObj.GetModuleDict()
- for ModKey in ModuleDict.keys():
- ModObj = ModuleDict[ModKey]
- if self.CheckModuleDepexSatisfied(ModObj, DpObj):
- continue
- else:
- return False
- return True
-
- ## Check whether a DP exists.
- #
- # @param Guid: Guid of a Distribution
- # @param Version: Version of a Distribution
- # @return: True if Distribution exist
- # False else
- def CheckDpExists(self, Guid, Version):
- Logger.Verbose(ST.MSG_CHECK_DP_START)
- DpList = self.IpiDb.GetDp(Guid, Version)
- if len(DpList) > 0:
- Found = True
- else:
- Found = False
-
- Logger.Verbose(ST.MSG_CHECK_DP_FINISH)
- return Found
-
- ## Check whether a DP depex satisfied by current workspace for Install
- #
- # @param DpObj: A distribution object
- # @return: True if distribution depex satisfied
- # False else
- #
- def CheckInstallDpDepexSatisfied(self, DpObj):
- return self.CheckDpDepexSatisfied(DpObj)
-
- # # Check whether multiple DP depex satisfied by current workspace for Install
- #
- # @param DpObjList: A distribution object list
- # @return: True if distribution depex satisfied
- # False else
- #
- def CheckTestInstallPdDepexSatisfied(self, DpObjList):
- for DpObj in DpObjList:
- if self.CheckDpDepexSatisfied(DpObj):
- for PkgKey in DpObj.PackageSurfaceArea.keys():
- PkgObj = DpObj.PackageSurfaceArea[PkgKey]
- self.PkgsToBeDepend.append((PkgObj.Guid, PkgObj.Version))
- else:
- return False, DpObj
-
- return True, DpObj
-
-
- ## Check whether a DP depex satisfied by current workspace
- # (excluding the original distribution's packages to be replaced) for Replace
- #
- # @param DpObj: A distribution object
- # @param OrigDpGuid: The original distribution's Guid
- # @param OrigDpVersion: The original distribution's Version
- #
- def ReplaceCheckNewDpDepex(self, DpObj, OrigDpGuid, OrigDpVersion):
- self.PkgsToBeDepend = [(PkgInfo[1], PkgInfo[2]) for PkgInfo in self.WsPkgList]
- OrigDpPackageList = self.IpiDb.GetPackageListFromDp(OrigDpGuid, OrigDpVersion)
- for OrigPkgInfo in OrigDpPackageList:
- Guid, Version = OrigPkgInfo[0], OrigPkgInfo[1]
- if (Guid, Version) in self.PkgsToBeDepend:
- self.PkgsToBeDepend.remove((Guid, Version))
- return self.CheckDpDepexSatisfied(DpObj)
-
- ## Check whether a DP depex satisfied by current workspace.
- #
- # @param DpObj: A distribution object
- #
- def CheckDpDepexSatisfied(self, DpObj):
- for PkgKey in DpObj.PackageSurfaceArea.keys():
- PkgObj = DpObj.PackageSurfaceArea[PkgKey]
- if self.CheckPackageDepexSatisfied(PkgObj, DpObj):
- continue
- else:
- return False
-
- for ModKey in DpObj.ModuleSurfaceArea.keys():
- ModObj = DpObj.ModuleSurfaceArea[ModKey]
- if self.CheckModuleDepexSatisfied(ModObj, DpObj):
- continue
- else:
- return False
-
- return True
-
- ## Check whether a DP could be removed from current workspace.
- #
- # @param DpGuid: File's guid
- # @param DpVersion: File's version
- # @retval Removable: True if distribution could be removed, False Else
- # @retval DependModuleList: the list of modules that make distribution can not be removed
- #
- def CheckDpDepexForRemove(self, DpGuid, DpVersion):
- Removable = True
- DependModuleList = []
- WsModuleList = self.WsModuleList
- #
- # remove modules that included in current DP
- # List of item (FilePath)
- DpModuleList = self.IpiDb.GetDpModuleList(DpGuid, DpVersion)
- for Module in DpModuleList:
- if Module in WsModuleList:
- WsModuleList.remove(Module)
- else:
- Logger.Warn("UPT\n",
- ST.ERR_MODULE_NOT_INSTALLED % Module)
- #
- # get packages in current Dp and find the install path
- # List of item (PkgGuid, PkgVersion, InstallPath)
- DpPackageList = self.IpiDb.GetPackageListFromDp(DpGuid, DpVersion)
- DpPackagePathList = []
- WorkSP = GlobalData.gWORKSPACE
- for (PkgName, PkgGuid, PkgVersion, DecFile) in self.WsPkgList:
- if PkgName:
- pass
- DecPath = dirname(DecFile)
- if DecPath.find(WorkSP) > -1:
- InstallPath = GetRelativePath(DecPath, WorkSP)
- DecFileRelaPath = GetRelativePath(DecFile, WorkSP)
- else:
- InstallPath = DecPath
- DecFileRelaPath = DecFile
-
- if (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
- DpPackagePathList.append(DecFileRelaPath)
- DpPackageList.remove((PkgGuid, PkgVersion, InstallPath))
-
- #
- # the left items in DpPackageList are the packages that installed but not found anymore
- #
- for (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
- Logger.Warn("UPT",
- ST.WARN_INSTALLED_PACKAGE_NOT_FOUND%(PkgGuid, PkgVersion, InstallPath))
-
- #
- # check modules to see if has dependency on package of current DP
- #
- for Module in WsModuleList:
- if (not VerifyRemoveModuleDep(Module, DpPackagePathList)):
- Removable = False
- DependModuleList.append(Module)
- return (Removable, DependModuleList)
-
-
- ## Check whether a DP could be replaced by a distribution containing NewDpPkgList
- # from current workspace.
- #
- # @param OrigDpGuid: original Dp's Guid
- # @param OrigDpVersion: original Dp's version
- # @param NewDpPkgList: a list of package information (Guid, Version) in new Dp
- # @retval Replaceable: True if distribution could be replaced, False Else
- # @retval DependModuleList: the list of modules that make distribution can not be replaced
- #
- def CheckDpDepexForReplace(self, OrigDpGuid, OrigDpVersion, NewDpPkgList):
- Replaceable = True
- DependModuleList = []
- WsModuleList = self.WsModuleList
- #
- # remove modules that included in current DP
- # List of item (FilePath)
- DpModuleList = self.IpiDb.GetDpModuleList(OrigDpGuid, OrigDpVersion)
- for Module in DpModuleList:
- if Module in WsModuleList:
- WsModuleList.remove(Module)
- else:
- Logger.Warn("UPT\n",
- ST.ERR_MODULE_NOT_INSTALLED % Module)
-
- OtherPkgList = NewDpPkgList
- #
- # get packages in current Dp and find the install path
- # List of item (PkgGuid, PkgVersion, InstallPath)
- DpPackageList = self.IpiDb.GetPackageListFromDp(OrigDpGuid, OrigDpVersion)
- DpPackagePathList = []
- WorkSP = GlobalData.gWORKSPACE
- for (PkgName, PkgGuid, PkgVersion, DecFile) in self.WsPkgList:
- if PkgName:
- pass
- DecPath = dirname(DecFile)
- if DecPath.find(WorkSP) > -1:
- InstallPath = GetRelativePath(DecPath, WorkSP)
- DecFileRelaPath = GetRelativePath(DecFile, WorkSP)
- else:
- InstallPath = DecPath
- DecFileRelaPath = DecFile
-
- if (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
- DpPackagePathList.append(DecFileRelaPath)
- DpPackageList.remove((PkgGuid, PkgVersion, InstallPath))
- else:
- OtherPkgList.append((PkgGuid, PkgVersion))
-
- #
- # the left items in DpPackageList are the packages that installed but not found anymore
- #
- for (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
- Logger.Warn("UPT",
- ST.WARN_INSTALLED_PACKAGE_NOT_FOUND%(PkgGuid, PkgVersion, InstallPath))
-
- #
- # check modules to see if it can be satisfied by package not belong to removed DP
- #
- for Module in WsModuleList:
- if (not VerifyReplaceModuleDep(Module, DpPackagePathList, OtherPkgList)):
- Replaceable = False
- DependModuleList.append(Module)
- return (Replaceable, DependModuleList)
-
-
-## check whether module depends on packages in DpPackagePathList, return True
-# if found, False else
-#
-# @param Path: a module path
-# @param DpPackagePathList: a list of Package Paths
-# @retval: False: module depends on package in DpPackagePathList
-# True: module doesn't depend on package in DpPackagePathList
-#
-def VerifyRemoveModuleDep(Path, DpPackagePathList):
- try:
- for Item in GetPackagePath(Path):
- if Item in DpPackagePathList:
- DecPath = os.path.normpath(os.path.join(GlobalData.gWORKSPACE, Item))
- Logger.Info(ST.MSG_MODULE_DEPEND_ON % (Path, DecPath))
- return False
- else:
- return True
- except FatalError as ErrCode:
- if ErrCode.message == EDK1_INF_ERROR:
- Logger.Warn("UPT",
- ST.WRN_EDK1_INF_FOUND%Path)
- return True
- else:
- return True
-
-# # GetPackagePath
-#
-# Get Dependency package path from an Inf file path
-#
-def GetPackagePath(InfPath):
- PackagePath = []
- if os.path.exists(InfPath):
- FindSection = False
- for Line in open(InfPath).readlines():
- Line = Line.strip()
- if not Line:
- continue
- if Line.startswith('#'):
- continue
- if Line.startswith('[Packages') and Line.endswith(']'):
- FindSection = True
- continue
- if Line.startswith('[') and Line.endswith(']') and FindSection:
- break
- if FindSection:
- PackagePath.append(os.path.normpath(Line))
-
- return PackagePath
-
-## check whether module depends on packages in DpPackagePathList and can not be satisfied by OtherPkgList
-#
-# @param Path: a module path
-# @param DpPackagePathList: a list of Package Paths
-# @param OtherPkgList: a list of Package Information (Guid, Version)
-# @retval: False: module depends on package in DpPackagePathList and can not be satisfied by OtherPkgList
-# True: either module doesn't depend on DpPackagePathList or module depends on DpPackagePathList
-# but can be satisfied by OtherPkgList
-#
-def VerifyReplaceModuleDep(Path, DpPackagePathList, OtherPkgList):
- try:
- for Item in GetPackagePath(Path):
- if Item in DpPackagePathList:
- DecPath = os.path.normpath(os.path.join(GlobalData.gWORKSPACE, Item))
- Name, Guid, Version = GetPkgInfoFromDec(DecPath)
- if (Guid, Version) not in OtherPkgList:
- Logger.Info(ST.MSG_MODULE_DEPEND_ON % (Path, DecPath))
- return False
- else:
- return True
- except FatalError as ErrCode:
- if ErrCode.message == EDK1_INF_ERROR:
- Logger.Warn("UPT",
- ST.WRN_EDK1_INF_FOUND%Path)
- return True
- else:
- return True
+## @file
+# This file is for installed package information database operations
+#
+# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+#
+
+'''
+Dependency
+'''
+
+##
+# Import Modules
+#
+from os.path import dirname
+import os
+
+import Logger.Log as Logger
+from Logger import StringTable as ST
+from Library.Parsing import GetWorkspacePackage
+from Library.Parsing import GetWorkspaceModule
+from Library.Parsing import GetPkgInfoFromDec
+from Library.Misc import GetRelativePath
+from Library import GlobalData
+from Logger.ToolError import FatalError
+from Logger.ToolError import EDK1_INF_ERROR
+from Logger.ToolError import UNKNOWN_ERROR
+(DEPEX_CHECK_SUCCESS, DEPEX_CHECK_MODULE_NOT_FOUND, \
+DEPEX_CHECK_PACKAGE_NOT_FOUND, DEPEX_CHECK_DP_NOT_FOUND) = (0, 1, 2, 3)
+
+
+## DependencyRules
+#
+# This class represents the dependency rule check mechanism
+#
+# @param object: Inherited from object class
+#
+class DependencyRules(object):
+ def __init__(self, Datab, ToBeInstalledPkgList=None):
+ self.IpiDb = Datab
+ self.WsPkgList = GetWorkspacePackage()
+ self.WsModuleList = GetWorkspaceModule()
+
+ self.PkgsToBeDepend = [(PkgInfo[1], PkgInfo[2]) for PkgInfo in self.WsPkgList]
+
+ # Add package info from the DIST to be installed.
+ self.PkgsToBeDepend.extend(self.GenToBeInstalledPkgList(ToBeInstalledPkgList))
+
+ def GenToBeInstalledPkgList(self, ToBeInstalledPkgList):
+ if not ToBeInstalledPkgList:
+ return []
+ RtnList = []
+ for Dist in ToBeInstalledPkgList:
+ for Package in Dist.PackageSurfaceArea:
+ RtnList.append((Package[0], Package[1]))
+
+ return RtnList
+
+ ## Check whether a module exists by checking the Guid+Version+Name+Path combination
+ #
+ # @param Guid: Guid of a module
+ # @param Version: Version of a module
+ # @param Name: Name of a module
+ # @param Path: Path of a module
+ # @return: True if module existed, else False
+ #
+ def CheckModuleExists(self, Guid, Version, Name, Path):
+ Logger.Verbose(ST.MSG_CHECK_MODULE_EXIST)
+ ModuleList = self.IpiDb.GetModInPackage(Guid, Version, Name, Path)
+ ModuleList.extend(self.IpiDb.GetStandaloneModule(Guid, Version, Name, Path))
+ Logger.Verbose(ST.MSG_CHECK_MODULE_EXIST_FINISH)
+ if len(ModuleList) > 0:
+ return True
+ else:
+ return False
+
+ ## Check whether a module depex satisfied.
+ #
+ # @param ModuleObj: A module object
+ # @param DpObj: A distribution object
+ # @return: True if module depex satisfied
+ # False else
+ #
+ def CheckModuleDepexSatisfied(self, ModuleObj, DpObj=None):
+ Logger.Verbose(ST.MSG_CHECK_MODULE_DEPEX_START)
+ Result = True
+ Dep = None
+ if ModuleObj.GetPackageDependencyList():
+ Dep = ModuleObj.GetPackageDependencyList()[0]
+ for Dep in ModuleObj.GetPackageDependencyList():
+ #
+ # first check whether the dependency satisfied by current workspace
+ #
+ Exist = self.CheckPackageExists(Dep.GetGuid(), Dep.GetVersion())
+ #
+ # check whether satisfied by current distribution
+ #
+ if not Exist:
+ if DpObj is None:
+ Result = False
+ break
+ for GuidVerPair in DpObj.PackageSurfaceArea.keys():
+ if Dep.GetGuid() == GuidVerPair[0]:
+ if Dep.GetVersion() is None or \
+ len(Dep.GetVersion()) == 0:
+ Result = True
+ break
+ if Dep.GetVersion() == GuidVerPair[1]:
+ Result = True
+ break
+ else:
+ Result = False
+ break
+
+ if not Result:
+ Logger.Error("CheckModuleDepex", UNKNOWN_ERROR, \
+ ST.ERR_DEPENDENCY_NOT_MATCH % (ModuleObj.GetName(), \
+ Dep.GetPackageFilePath(), \
+ Dep.GetGuid(), \
+ Dep.GetVersion()))
+ return Result
+
+ ## Check whether a package exists in a package list specified by PkgsToBeDepend.
+ #
+ # @param Guid: Guid of a package
+ # @param Version: Version of a package
+ # @return: True if package exist
+ # False else
+ #
+ def CheckPackageExists(self, Guid, Version):
+ Logger.Verbose(ST.MSG_CHECK_PACKAGE_START)
+ Found = False
+ for (PkgGuid, PkgVer) in self.PkgsToBeDepend:
+ if (PkgGuid == Guid):
+ #
+ # if version is not empty and not equal, then not match
+ #
+ if Version and (PkgVer != Version):
+ Found = False
+ break
+ else:
+ Found = True
+ break
+ else:
+ Found = False
+
+ Logger.Verbose(ST.MSG_CHECK_PACKAGE_FINISH)
+ return Found
+
+ ## Check whether a package depex satisfied.
+ #
+ # @param PkgObj: A package object
+ # @param DpObj: A distribution object
+ # @return: True if package depex satisfied
+ # False else
+ #
+ def CheckPackageDepexSatisfied(self, PkgObj, DpObj=None):
+ ModuleDict = PkgObj.GetModuleDict()
+ for ModKey in ModuleDict.keys():
+ ModObj = ModuleDict[ModKey]
+ if self.CheckModuleDepexSatisfied(ModObj, DpObj):
+ continue
+ else:
+ return False
+ return True
+
+ ## Check whether a DP exists.
+ #
+ # @param Guid: Guid of a Distribution
+ # @param Version: Version of a Distribution
+ # @return: True if Distribution exist
+ # False else
+ def CheckDpExists(self, Guid, Version):
+ Logger.Verbose(ST.MSG_CHECK_DP_START)
+ DpList = self.IpiDb.GetDp(Guid, Version)
+ if len(DpList) > 0:
+ Found = True
+ else:
+ Found = False
+
+ Logger.Verbose(ST.MSG_CHECK_DP_FINISH)
+ return Found
+
+ ## Check whether a DP depex satisfied by current workspace for Install
+ #
+ # @param DpObj: A distribution object
+ # @return: True if distribution depex satisfied
+ # False else
+ #
+ def CheckInstallDpDepexSatisfied(self, DpObj):
+ return self.CheckDpDepexSatisfied(DpObj)
+
+ # # Check whether multiple DP depex satisfied by current workspace for Install
+ #
+ # @param DpObjList: A distribution object list
+ # @return: True if distribution depex satisfied
+ # False else
+ #
+ def CheckTestInstallPdDepexSatisfied(self, DpObjList):
+ for DpObj in DpObjList:
+ if self.CheckDpDepexSatisfied(DpObj):
+ for PkgKey in DpObj.PackageSurfaceArea.keys():
+ PkgObj = DpObj.PackageSurfaceArea[PkgKey]
+ self.PkgsToBeDepend.append((PkgObj.Guid, PkgObj.Version))
+ else:
+ return False, DpObj
+
+ return True, DpObj
+
+
+ ## Check whether a DP depex satisfied by current workspace
+ # (excluding the original distribution's packages to be replaced) for Replace
+ #
+ # @param DpObj: A distribution object
+ # @param OrigDpGuid: The original distribution's Guid
+ # @param OrigDpVersion: The original distribution's Version
+ #
+ def ReplaceCheckNewDpDepex(self, DpObj, OrigDpGuid, OrigDpVersion):
+ self.PkgsToBeDepend = [(PkgInfo[1], PkgInfo[2]) for PkgInfo in self.WsPkgList]
+ OrigDpPackageList = self.IpiDb.GetPackageListFromDp(OrigDpGuid, OrigDpVersion)
+ for OrigPkgInfo in OrigDpPackageList:
+ Guid, Version = OrigPkgInfo[0], OrigPkgInfo[1]
+ if (Guid, Version) in self.PkgsToBeDepend:
+ self.PkgsToBeDepend.remove((Guid, Version))
+ return self.CheckDpDepexSatisfied(DpObj)
+
+ ## Check whether a DP depex satisfied by current workspace.
+ #
+ # @param DpObj: A distribution object
+ #
+ def CheckDpDepexSatisfied(self, DpObj):
+ for PkgKey in DpObj.PackageSurfaceArea.keys():
+ PkgObj = DpObj.PackageSurfaceArea[PkgKey]
+ if self.CheckPackageDepexSatisfied(PkgObj, DpObj):
+ continue
+ else:
+ return False
+
+ for ModKey in DpObj.ModuleSurfaceArea.keys():
+ ModObj = DpObj.ModuleSurfaceArea[ModKey]
+ if self.CheckModuleDepexSatisfied(ModObj, DpObj):
+ continue
+ else:
+ return False
+
+ return True
+
+ ## Check whether a DP could be removed from current workspace.
+ #
+ # @param DpGuid: File's guid
+ # @param DpVersion: File's version
+ # @retval Removable: True if distribution could be removed, False Else
+ # @retval DependModuleList: the list of modules that make distribution can not be removed
+ #
+ def CheckDpDepexForRemove(self, DpGuid, DpVersion):
+ Removable = True
+ DependModuleList = []
+ WsModuleList = self.WsModuleList
+ #
+ # remove modules that included in current DP
+ # List of item (FilePath)
+ DpModuleList = self.IpiDb.GetDpModuleList(DpGuid, DpVersion)
+ for Module in DpModuleList:
+ if Module in WsModuleList:
+ WsModuleList.remove(Module)
+ else:
+ Logger.Warn("UPT\n",
+ ST.ERR_MODULE_NOT_INSTALLED % Module)
+ #
+ # get packages in current Dp and find the install path
+ # List of item (PkgGuid, PkgVersion, InstallPath)
+ DpPackageList = self.IpiDb.GetPackageListFromDp(DpGuid, DpVersion)
+ DpPackagePathList = []
+ WorkSP = GlobalData.gWORKSPACE
+ for (PkgName, PkgGuid, PkgVersion, DecFile) in self.WsPkgList:
+ if PkgName:
+ pass
+ DecPath = dirname(DecFile)
+ if DecPath.find(WorkSP) > -1:
+ InstallPath = GetRelativePath(DecPath, WorkSP)
+ DecFileRelaPath = GetRelativePath(DecFile, WorkSP)
+ else:
+ InstallPath = DecPath
+ DecFileRelaPath = DecFile
+
+ if (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
+ DpPackagePathList.append(DecFileRelaPath)
+ DpPackageList.remove((PkgGuid, PkgVersion, InstallPath))
+
+ #
+ # the left items in DpPackageList are the packages that installed but not found anymore
+ #
+ for (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
+ Logger.Warn("UPT",
+ ST.WARN_INSTALLED_PACKAGE_NOT_FOUND%(PkgGuid, PkgVersion, InstallPath))
+
+ #
+ # check modules to see if has dependency on package of current DP
+ #
+ for Module in WsModuleList:
+ if (not VerifyRemoveModuleDep(Module, DpPackagePathList)):
+ Removable = False
+ DependModuleList.append(Module)
+ return (Removable, DependModuleList)
+
+
+ ## Check whether a DP could be replaced by a distribution containing NewDpPkgList
+ # from current workspace.
+ #
+ # @param OrigDpGuid: original Dp's Guid
+ # @param OrigDpVersion: original Dp's version
+ # @param NewDpPkgList: a list of package information (Guid, Version) in new Dp
+ # @retval Replaceable: True if distribution could be replaced, False Else
+ # @retval DependModuleList: the list of modules that make distribution can not be replaced
+ #
+ def CheckDpDepexForReplace(self, OrigDpGuid, OrigDpVersion, NewDpPkgList):
+ Replaceable = True
+ DependModuleList = []
+ WsModuleList = self.WsModuleList
+ #
+ # remove modules that included in current DP
+ # List of item (FilePath)
+ DpModuleList = self.IpiDb.GetDpModuleList(OrigDpGuid, OrigDpVersion)
+ for Module in DpModuleList:
+ if Module in WsModuleList:
+ WsModuleList.remove(Module)
+ else:
+ Logger.Warn("UPT\n",
+ ST.ERR_MODULE_NOT_INSTALLED % Module)
+
+ OtherPkgList = NewDpPkgList
+ #
+ # get packages in current Dp and find the install path
+ # List of item (PkgGuid, PkgVersion, InstallPath)
+ DpPackageList = self.IpiDb.GetPackageListFromDp(OrigDpGuid, OrigDpVersion)
+ DpPackagePathList = []
+ WorkSP = GlobalData.gWORKSPACE
+ for (PkgName, PkgGuid, PkgVersion, DecFile) in self.WsPkgList:
+ if PkgName:
+ pass
+ DecPath = dirname(DecFile)
+ if DecPath.find(WorkSP) > -1:
+ InstallPath = GetRelativePath(DecPath, WorkSP)
+ DecFileRelaPath = GetRelativePath(DecFile, WorkSP)
+ else:
+ InstallPath = DecPath
+ DecFileRelaPath = DecFile
+
+ if (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
+ DpPackagePathList.append(DecFileRelaPath)
+ DpPackageList.remove((PkgGuid, PkgVersion, InstallPath))
+ else:
+ OtherPkgList.append((PkgGuid, PkgVersion))
+
+ #
+ # the left items in DpPackageList are the packages that installed but not found anymore
+ #
+ for (PkgGuid, PkgVersion, InstallPath) in DpPackageList:
+ Logger.Warn("UPT",
+ ST.WARN_INSTALLED_PACKAGE_NOT_FOUND%(PkgGuid, PkgVersion, InstallPath))
+
+ #
+ # check modules to see if it can be satisfied by package not belong to removed DP
+ #
+ for Module in WsModuleList:
+ if (not VerifyReplaceModuleDep(Module, DpPackagePathList, OtherPkgList)):
+ Replaceable = False
+ DependModuleList.append(Module)
+ return (Replaceable, DependModuleList)
+
+
+## check whether module depends on packages in DpPackagePathList, return True
+# if found, False else
+#
+# @param Path: a module path
+# @param DpPackagePathList: a list of Package Paths
+# @retval: False: module depends on package in DpPackagePathList
+# True: module doesn't depend on package in DpPackagePathList
+#
+def VerifyRemoveModuleDep(Path, DpPackagePathList):
+ try:
+ for Item in GetPackagePath(Path):
+ if Item in DpPackagePathList:
+ DecPath = os.path.normpath(os.path.join(GlobalData.gWORKSPACE, Item))
+ Logger.Info(ST.MSG_MODULE_DEPEND_ON % (Path, DecPath))
+ return False
+ else:
+ return True
+ except FatalError as ErrCode:
+ if ErrCode.message == EDK1_INF_ERROR:
+ Logger.Warn("UPT",
+ ST.WRN_EDK1_INF_FOUND%Path)
+ return True
+ else:
+ return True
+
+# # GetPackagePath
+#
+# Get Dependency package path from an Inf file path
+#
+def GetPackagePath(InfPath):
+ PackagePath = []
+ if os.path.exists(InfPath):
+ FindSection = False
+ for Line in open(InfPath).readlines():
+ Line = Line.strip()
+ if not Line:
+ continue
+ if Line.startswith('#'):
+ continue
+ if Line.startswith('[Packages') and Line.endswith(']'):
+ FindSection = True
+ continue
+ if Line.startswith('[') and Line.endswith(']') and FindSection:
+ break
+ if FindSection:
+ PackagePath.append(os.path.normpath(Line))
+
+ return PackagePath
+
+## check whether module depends on packages in DpPackagePathList and can not be satisfied by OtherPkgList
+#
+# @param Path: a module path
+# @param DpPackagePathList: a list of Package Paths
+# @param OtherPkgList: a list of Package Information (Guid, Version)
+# @retval: False: module depends on package in DpPackagePathList and can not be satisfied by OtherPkgList
+# True: either module doesn't depend on DpPackagePathList or module depends on DpPackagePathList
+# but can be satisfied by OtherPkgList
+#
+def VerifyReplaceModuleDep(Path, DpPackagePathList, OtherPkgList):
+ try:
+ for Item in GetPackagePath(Path):
+ if Item in DpPackagePathList:
+ DecPath = os.path.normpath(os.path.join(GlobalData.gWORKSPACE, Item))
+ Name, Guid, Version = GetPkgInfoFromDec(DecPath)
+ if (Guid, Version) not in OtherPkgList:
+ Logger.Info(ST.MSG_MODULE_DEPEND_ON % (Path, DecPath))
+ return False
+ else:
+ return True
+ except FatalError as ErrCode:
+ if ErrCode.message == EDK1_INF_ERROR:
+ Logger.Warn("UPT",
+ ST.WRN_EDK1_INF_FOUND%Path)
+ return True
+ else:
+ return True
diff --git a/BaseTools/Source/Python/UPT/Core/DistributionPackageClass.py b/BaseTools/Source/Python/UPT/Core/DistributionPackageClass.py
index 5ee79b6317..c9b3ccbc79 100644
--- a/BaseTools/Source/Python/UPT/Core/DistributionPackageClass.py
+++ b/BaseTools/Source/Python/UPT/Core/DistributionPackageClass.py
@@ -1,267 +1,267 @@
-## @file
-# This file is used to define a class object to describe a distribution package
-#
-# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-
-'''
-DistributionPackageClass
-'''
-
-##
-# Import Modules
-#
-import os.path
-
-from Library.Misc import Sdict
-from Library.Misc import GetNonMetaDataFiles
-from PomAdapter.InfPomAlignment import InfPomAlignment
-from PomAdapter.DecPomAlignment import DecPomAlignment
-import Logger.Log as Logger
-from Logger import StringTable as ST
-from Logger.ToolError import OPTION_VALUE_INVALID
-from Logger.ToolError import FatalError
-from Logger.ToolError import EDK1_INF_ERROR
-from Object.POM.CommonObject import IdentificationObject
-from Object.POM.CommonObject import CommonHeaderObject
-from Object.POM.CommonObject import MiscFileObject
-from Common.MultipleWorkspace import MultipleWorkspace as mws
-
-## DistributionPackageHeaderClass
-#
-# @param IdentificationObject: Identification Object
-# @param CommonHeaderObject: Common Header Object
-#
-class DistributionPackageHeaderObject(IdentificationObject, \
- CommonHeaderObject):
- def __init__(self):
- IdentificationObject.__init__(self)
- CommonHeaderObject.__init__(self)
- self.ReadOnly = ''
- self.RePackage = ''
- self.Vendor = ''
- self.Date = ''
- self.Signature = 'Md5Sum'
- self.XmlSpecification = ''
-
- def GetReadOnly(self):
- return self.ReadOnly
-
- def SetReadOnly(self, ReadOnly):
- self.ReadOnly = ReadOnly
-
- def GetRePackage(self):
- return self.RePackage
-
- def SetRePackage(self, RePackage):
- self.RePackage = RePackage
-
- def GetVendor(self):
- return self.Vendor
-
- def SetDate(self, Date):
- self.Date = Date
-
- def GetDate(self):
- return self.Date
-
- def SetSignature(self, Signature):
- self.Signature = Signature
-
- def GetSignature(self):
- return self.Signature
-
- def SetXmlSpecification(self, XmlSpecification):
- self.XmlSpecification = XmlSpecification
-
- def GetXmlSpecification(self):
- return self.XmlSpecification
-
-## DistributionPackageClass
-#
-# @param object: DistributionPackageClass
-#
-class DistributionPackageClass(object):
- def __init__(self):
- self.Header = DistributionPackageHeaderObject()
- #
- # {(Guid, Version, Path) : PackageObj}
- #
- self.PackageSurfaceArea = Sdict()
- #
- # {(Guid, Version, Name, Path) : ModuleObj}
- #
- self.ModuleSurfaceArea = Sdict()
- self.Tools = MiscFileObject()
- self.MiscellaneousFiles = MiscFileObject()
- self.UserExtensions = []
- self.FileList = []
-
- ## Get all included packages and modules for a distribution package
- #
- # @param WorkspaceDir: WorkspaceDir
- # @param PackageList: A list of all packages
- # @param ModuleList: A list of all modules
- #
- def GetDistributionPackage(self, WorkspaceDir, PackageList, ModuleList):
- # Backup WorkspaceDir
- Root = WorkspaceDir
-
- #
- # Get Packages
- #
- if PackageList:
- for PackageFile in PackageList:
- PackageFileFullPath = mws.join(Root, PackageFile)
- WorkspaceDir = mws.getWs(Root, PackageFile)
- DecObj = DecPomAlignment(PackageFileFullPath, WorkspaceDir, CheckMulDec=True)
- PackageObj = DecObj
- #
- # Parser inf file one bye one
- #
- ModuleInfFileList = PackageObj.GetModuleFileList()
- for File in ModuleInfFileList:
- WsRelPath = os.path.join(PackageObj.GetPackagePath(), File)
- WsRelPath = os.path.normpath(WsRelPath)
- if ModuleList and WsRelPath in ModuleList:
- Logger.Error("UPT",
- OPTION_VALUE_INVALID,
- ST.ERR_NOT_STANDALONE_MODULE_ERROR%\
- (WsRelPath, PackageFile))
- Filename = os.path.normpath\
- (os.path.join(PackageObj.GetRelaPath(), File))
- os.path.splitext(Filename)
- #
- # Call INF parser to generate Inf Object.
- # Actually, this call is not directly call, but wrapped by
- # Inf class in InfPomAlignment.
- #
- try:
- ModuleObj = InfPomAlignment(Filename, WorkspaceDir, PackageObj.GetPackagePath())
-
- #
- # Add module to package
- #
- ModuleDict = PackageObj.GetModuleDict()
- ModuleDict[(ModuleObj.GetGuid(), \
- ModuleObj.GetVersion(), \
- ModuleObj.GetName(), \
- ModuleObj.GetCombinePath())] = ModuleObj
- PackageObj.SetModuleDict(ModuleDict)
- except FatalError as ErrCode:
- if ErrCode.message == EDK1_INF_ERROR:
- Logger.Warn("UPT",
- ST.WRN_EDK1_INF_FOUND%Filename)
- else:
- raise
-
- self.PackageSurfaceArea\
- [(PackageObj.GetGuid(), PackageObj.GetVersion(), \
- PackageObj.GetCombinePath())] = PackageObj
-
- #
- # Get Modules
- #
- if ModuleList:
- for ModuleFile in ModuleList:
- ModuleFileFullPath = mws.join(Root, ModuleFile)
- WorkspaceDir = mws.getWs(Root, ModuleFile)
-
- try:
- ModuleObj = InfPomAlignment(ModuleFileFullPath, WorkspaceDir)
- ModuleKey = (ModuleObj.GetGuid(),
- ModuleObj.GetVersion(),
- ModuleObj.GetName(),
- ModuleObj.GetCombinePath())
- self.ModuleSurfaceArea[ModuleKey] = ModuleObj
- except FatalError as ErrCode:
- if ErrCode.message == EDK1_INF_ERROR:
- Logger.Error("UPT",
- EDK1_INF_ERROR,
- ST.WRN_EDK1_INF_FOUND%ModuleFileFullPath,
- ExtraData=ST.ERR_NOT_SUPPORTED_SA_MODULE)
- else:
- raise
-
- # Recover WorkspaceDir
- WorkspaceDir = Root
-
- ## Get all files included for a distribution package, except tool/misc of
- # distribution level
- #
- # @retval DistFileList A list of filepath for NonMetaDataFile, relative to workspace
- # @retval MetaDataFileList A list of filepath for MetaDataFile, relative to workspace
- #
- def GetDistributionFileList(self):
- MetaDataFileList = []
- SkipModulesUniList = []
-
- for Guid, Version, Path in self.PackageSurfaceArea:
- Package = self.PackageSurfaceArea[Guid, Version, Path]
- PackagePath = Package.GetPackagePath()
- FullPath = Package.GetFullPath()
- MetaDataFileList.append(Path)
- IncludePathList = Package.GetIncludePathList()
- for IncludePath in IncludePathList:
- SearchPath = os.path.normpath(os.path.join(os.path.dirname(FullPath), IncludePath))
- AddPath = os.path.normpath(os.path.join(PackagePath, IncludePath))
- self.FileList += GetNonMetaDataFiles(SearchPath, ['CVS', '.svn'], False, AddPath)
- #
- # Add the miscellaneous files on DEC file
- #
- for MiscFileObj in Package.GetMiscFileList():
- for FileObj in MiscFileObj.GetFileList():
- MiscFileFullPath = os.path.normpath(os.path.join(PackagePath, FileObj.GetURI()))
- if MiscFileFullPath not in self.FileList:
- self.FileList.append(MiscFileFullPath)
-
- Module = None
- ModuleDict = Package.GetModuleDict()
- for Guid, Version, Name, Path in ModuleDict:
- Module = ModuleDict[Guid, Version, Name, Path]
- ModulePath = Module.GetModulePath()
- FullPath = Module.GetFullPath()
- PkgRelPath = os.path.normpath(os.path.join(PackagePath, ModulePath))
- MetaDataFileList.append(Path)
- SkipList = ['CVS', '.svn']
- NonMetaDataFileList = []
- if Module.UniFileClassObject:
- for UniFile in Module.UniFileClassObject.IncFileList:
- OriPath = os.path.normpath(os.path.dirname(FullPath))
- UniFilePath = os.path.normpath(os.path.join(PkgRelPath, UniFile.Path[len(OriPath) + 1:]))
- if UniFilePath not in SkipModulesUniList:
- SkipModulesUniList.append(UniFilePath)
- for IncludeFile in Module.UniFileClassObject.IncludePathList:
- if IncludeFile not in SkipModulesUniList:
- SkipModulesUniList.append(IncludeFile)
- NonMetaDataFileList = GetNonMetaDataFiles(os.path.dirname(FullPath), SkipList, False, PkgRelPath)
- for NonMetaDataFile in NonMetaDataFileList:
- if NonMetaDataFile not in self.FileList:
- self.FileList.append(NonMetaDataFile)
- for Guid, Version, Name, Path in self.ModuleSurfaceArea:
- Module = self.ModuleSurfaceArea[Guid, Version, Name, Path]
- ModulePath = Module.GetModulePath()
- FullPath = Module.GetFullPath()
- MetaDataFileList.append(Path)
- SkipList = ['CVS', '.svn']
- NonMetaDataFileList = []
- if Module.UniFileClassObject:
- for UniFile in Module.UniFileClassObject.IncFileList:
- OriPath = os.path.normpath(os.path.dirname(FullPath))
- UniFilePath = os.path.normpath(os.path.join(ModulePath, UniFile.Path[len(OriPath) + 1:]))
- if UniFilePath not in SkipModulesUniList:
- SkipModulesUniList.append(UniFilePath)
- NonMetaDataFileList = GetNonMetaDataFiles(os.path.dirname(FullPath), SkipList, False, ModulePath)
- for NonMetaDataFile in NonMetaDataFileList:
- if NonMetaDataFile not in self.FileList:
- self.FileList.append(NonMetaDataFile)
-
- for SkipModuleUni in SkipModulesUniList:
- if SkipModuleUni in self.FileList:
- self.FileList.remove(SkipModuleUni)
-
- return self.FileList, MetaDataFileList
-
-
-
+## @file
+# This file is used to define a class object to describe a distribution package
+#
+# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+
+'''
+DistributionPackageClass
+'''
+
+##
+# Import Modules
+#
+import os.path
+
+from Library.Misc import Sdict
+from Library.Misc import GetNonMetaDataFiles
+from PomAdapter.InfPomAlignment import InfPomAlignment
+from PomAdapter.DecPomAlignment import DecPomAlignment
+import Logger.Log as Logger
+from Logger import StringTable as ST
+from Logger.ToolError import OPTION_VALUE_INVALID
+from Logger.ToolError import FatalError
+from Logger.ToolError import EDK1_INF_ERROR
+from Object.POM.CommonObject import IdentificationObject
+from Object.POM.CommonObject import CommonHeaderObject
+from Object.POM.CommonObject import MiscFileObject
+from Common.MultipleWorkspace import MultipleWorkspace as mws
+
+## DistributionPackageHeaderClass
+#
+# @param IdentificationObject: Identification Object
+# @param CommonHeaderObject: Common Header Object
+#
+class DistributionPackageHeaderObject(IdentificationObject, \
+ CommonHeaderObject):
+ def __init__(self):
+ IdentificationObject.__init__(self)
+ CommonHeaderObject.__init__(self)
+ self.ReadOnly = ''
+ self.RePackage = ''
+ self.Vendor = ''
+ self.Date = ''
+ self.Signature = 'Md5Sum'
+ self.XmlSpecification = ''
+
+ def GetReadOnly(self):
+ return self.ReadOnly
+
+ def SetReadOnly(self, ReadOnly):
+ self.ReadOnly = ReadOnly
+
+ def GetRePackage(self):
+ return self.RePackage
+
+ def SetRePackage(self, RePackage):
+ self.RePackage = RePackage
+
+ def GetVendor(self):
+ return self.Vendor
+
+ def SetDate(self, Date):
+ self.Date = Date
+
+ def GetDate(self):
+ return self.Date
+
+ def SetSignature(self, Signature):
+ self.Signature = Signature
+
+ def GetSignature(self):
+ return self.Signature
+
+ def SetXmlSpecification(self, XmlSpecification):
+ self.XmlSpecification = XmlSpecification
+
+ def GetXmlSpecification(self):
+ return self.XmlSpecification
+
+## DistributionPackageClass
+#
+# @param object: DistributionPackageClass
+#
+class DistributionPackageClass(object):
+ def __init__(self):
+ self.Header = DistributionPackageHeaderObject()
+ #
+ # {(Guid, Version, Path) : PackageObj}
+ #
+ self.PackageSurfaceArea = Sdict()
+ #
+ # {(Guid, Version, Name, Path) : ModuleObj}
+ #
+ self.ModuleSurfaceArea = Sdict()
+ self.Tools = MiscFileObject()
+ self.MiscellaneousFiles = MiscFileObject()
+ self.UserExtensions = []
+ self.FileList = []
+
+ ## Get all included packages and modules for a distribution package
+ #
+ # @param WorkspaceDir: WorkspaceDir
+ # @param PackageList: A list of all packages
+ # @param ModuleList: A list of all modules
+ #
+ def GetDistributionPackage(self, WorkspaceDir, PackageList, ModuleList):
+ # Backup WorkspaceDir
+ Root = WorkspaceDir
+
+ #
+ # Get Packages
+ #
+ if PackageList:
+ for PackageFile in PackageList:
+ PackageFileFullPath = mws.join(Root, PackageFile)
+ WorkspaceDir = mws.getWs(Root, PackageFile)
+ DecObj = DecPomAlignment(PackageFileFullPath, WorkspaceDir, CheckMulDec=True)
+ PackageObj = DecObj
+ #
+ # Parser inf file one bye one
+ #
+ ModuleInfFileList = PackageObj.GetModuleFileList()
+ for File in ModuleInfFileList:
+ WsRelPath = os.path.join(PackageObj.GetPackagePath(), File)
+ WsRelPath = os.path.normpath(WsRelPath)
+ if ModuleList and WsRelPath in ModuleList:
+ Logger.Error("UPT",
+ OPTION_VALUE_INVALID,
+ ST.ERR_NOT_STANDALONE_MODULE_ERROR%\
+ (WsRelPath, PackageFile))
+ Filename = os.path.normpath\
+ (os.path.join(PackageObj.GetRelaPath(), File))
+ os.path.splitext(Filename)
+ #
+ # Call INF parser to generate Inf Object.
+ # Actually, this call is not directly call, but wrapped by
+ # Inf class in InfPomAlignment.
+ #
+ try:
+ ModuleObj = InfPomAlignment(Filename, WorkspaceDir, PackageObj.GetPackagePath())
+
+ #
+ # Add module to package
+ #
+ ModuleDict = PackageObj.GetModuleDict()
+ ModuleDict[(ModuleObj.GetGuid(), \
+ ModuleObj.GetVersion(), \
+ ModuleObj.GetName(), \
+ ModuleObj.GetCombinePath())] = ModuleObj
+ PackageObj.SetModuleDict(ModuleDict)
+ except FatalError as ErrCode:
+ if ErrCode.message == EDK1_INF_ERROR:
+ Logger.Warn("UPT",
+ ST.WRN_EDK1_INF_FOUND%Filename)
+ else:
+ raise
+
+ self.PackageSurfaceArea\
+ [(PackageObj.GetGuid(), PackageObj.GetVersion(), \
+ PackageObj.GetCombinePath())] = PackageObj
+
+ #
+ # Get Modules
+ #
+ if ModuleList:
+ for ModuleFile in ModuleList:
+ ModuleFileFullPath = mws.join(Root, ModuleFile)
+ WorkspaceDir = mws.getWs(Root, ModuleFile)
+
+ try:
+ ModuleObj = InfPomAlignment(ModuleFileFullPath, WorkspaceDir)
+ ModuleKey = (ModuleObj.GetGuid(),
+ ModuleObj.GetVersion(),
+ ModuleObj.GetName(),
+ ModuleObj.GetCombinePath())
+ self.ModuleSurfaceArea[ModuleKey] = ModuleObj
+ except FatalError as ErrCode:
+ if ErrCode.message == EDK1_INF_ERROR:
+ Logger.Error("UPT",
+ EDK1_INF_ERROR,
+ ST.WRN_EDK1_INF_FOUND%ModuleFileFullPath,
+ ExtraData=ST.ERR_NOT_SUPPORTED_SA_MODULE)
+ else:
+ raise
+
+ # Recover WorkspaceDir
+ WorkspaceDir = Root
+
+ ## Get all files included for a distribution package, except tool/misc of
+ # distribution level
+ #
+ # @retval DistFileList A list of filepath for NonMetaDataFile, relative to workspace
+ # @retval MetaDataFileList A list of filepath for MetaDataFile, relative to workspace
+ #
+ def GetDistributionFileList(self):
+ MetaDataFileList = []
+ SkipModulesUniList = []
+
+ for Guid, Version, Path in self.PackageSurfaceArea:
+ Package = self.PackageSurfaceArea[Guid, Version, Path]
+ PackagePath = Package.GetPackagePath()
+ FullPath = Package.GetFullPath()
+ MetaDataFileList.append(Path)
+ IncludePathList = Package.GetIncludePathList()
+ for IncludePath in IncludePathList:
+ SearchPath = os.path.normpath(os.path.join(os.path.dirname(FullPath), IncludePath))
+ AddPath = os.path.normpath(os.path.join(PackagePath, IncludePath))
+ self.FileList += GetNonMetaDataFiles(SearchPath, ['CVS', '.svn'], False, AddPath)
+ #
+ # Add the miscellaneous files on DEC file
+ #
+ for MiscFileObj in Package.GetMiscFileList():
+ for FileObj in MiscFileObj.GetFileList():
+ MiscFileFullPath = os.path.normpath(os.path.join(PackagePath, FileObj.GetURI()))
+ if MiscFileFullPath not in self.FileList:
+ self.FileList.append(MiscFileFullPath)
+
+ Module = None
+ ModuleDict = Package.GetModuleDict()
+ for Guid, Version, Name, Path in ModuleDict:
+ Module = ModuleDict[Guid, Version, Name, Path]
+ ModulePath = Module.GetModulePath()
+ FullPath = Module.GetFullPath()
+ PkgRelPath = os.path.normpath(os.path.join(PackagePath, ModulePath))
+ MetaDataFileList.append(Path)
+ SkipList = ['CVS', '.svn']
+ NonMetaDataFileList = []
+ if Module.UniFileClassObject:
+ for UniFile in Module.UniFileClassObject.IncFileList:
+ OriPath = os.path.normpath(os.path.dirname(FullPath))
+ UniFilePath = os.path.normpath(os.path.join(PkgRelPath, UniFile.Path[len(OriPath) + 1:]))
+ if UniFilePath not in SkipModulesUniList:
+ SkipModulesUniList.append(UniFilePath)
+ for IncludeFile in Module.UniFileClassObject.IncludePathList:
+ if IncludeFile not in SkipModulesUniList:
+ SkipModulesUniList.append(IncludeFile)
+ NonMetaDataFileList = GetNonMetaDataFiles(os.path.dirname(FullPath), SkipList, False, PkgRelPath)
+ for NonMetaDataFile in NonMetaDataFileList:
+ if NonMetaDataFile not in self.FileList:
+ self.FileList.append(NonMetaDataFile)
+ for Guid, Version, Name, Path in self.ModuleSurfaceArea:
+ Module = self.ModuleSurfaceArea[Guid, Version, Name, Path]
+ ModulePath = Module.GetModulePath()
+ FullPath = Module.GetFullPath()
+ MetaDataFileList.append(Path)
+ SkipList = ['CVS', '.svn']
+ NonMetaDataFileList = []
+ if Module.UniFileClassObject:
+ for UniFile in Module.UniFileClassObject.IncFileList:
+ OriPath = os.path.normpath(os.path.dirname(FullPath))
+ UniFilePath = os.path.normpath(os.path.join(ModulePath, UniFile.Path[len(OriPath) + 1:]))
+ if UniFilePath not in SkipModulesUniList:
+ SkipModulesUniList.append(UniFilePath)
+ NonMetaDataFileList = GetNonMetaDataFiles(os.path.dirname(FullPath), SkipList, False, ModulePath)
+ for NonMetaDataFile in NonMetaDataFileList:
+ if NonMetaDataFile not in self.FileList:
+ self.FileList.append(NonMetaDataFile)
+
+ for SkipModuleUni in SkipModulesUniList:
+ if SkipModuleUni in self.FileList:
+ self.FileList.remove(SkipModuleUni)
+
+ return self.FileList, MetaDataFileList
+
+
+
diff --git a/BaseTools/Source/Python/UPT/Core/FileHook.py b/BaseTools/Source/Python/UPT/Core/FileHook.py
index 20712065f7..433b4e3bd0 100644
--- a/BaseTools/Source/Python/UPT/Core/FileHook.py
+++ b/BaseTools/Source/Python/UPT/Core/FileHook.py
@@ -1,193 +1,193 @@
-## @file
-# This file hooks file and directory creation and removal
-#
-# Copyright (c) 2014 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-#
-
-'''
-File hook
-'''
-
-import os
-import stat
-import time
-import zipfile
-from time import sleep
-from Library import GlobalData
-
-__built_in_remove__ = os.remove
-__built_in_mkdir__ = os.mkdir
-__built_in_rmdir__ = os.rmdir
-__built_in_chmod__ = os.chmod
-__built_in_open__ = open
-
-_RMFILE = 0
-_MKFILE = 1
-_RMDIR = 2
-_MKDIR = 3
-_CHMOD = 4
-
-gBACKUPFILE = 'file.backup'
-gEXCEPTION_LIST = ['Conf'+os.sep+'DistributionPackageDatabase.db', '.tmp', gBACKUPFILE]
-
-class _PathInfo:
- def __init__(self, action, path, mode=-1):
- self.action = action
- self.path = path
- self.mode = mode
-
-class RecoverMgr:
- def __init__(self, workspace):
- self.rlist = []
- self.zip = None
- self.workspace = os.path.normpath(workspace)
- self.backupfile = gBACKUPFILE
- self.zipfile = os.path.join(self.workspace, gBACKUPFILE)
-
- def _createzip(self):
- if self.zip:
- return
- self.zip = zipfile.ZipFile(self.zipfile, 'w', zipfile.ZIP_DEFLATED)
-
- def _save(self, tmp, path):
- if not self._tryhook(path):
- return
- self.rlist.append(_PathInfo(tmp, path))
-
- def bkrmfile(self, path):
- arc = self._tryhook(path)
- if arc and os.path.isfile(path):
- self._createzip()
- self.zip.write(path, arc.encode('utf_8'))
- sta = os.stat(path)
- oldmode = stat.S_IMODE(sta.st_mode)
- self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
- self.rlist.append(_PathInfo(_RMFILE, path))
- __built_in_remove__(path)
-
- def bkmkfile(self, path, mode, bufsize):
- if not os.path.exists(path):
- self._save(_MKFILE, path)
- return __built_in_open__(path, mode, bufsize)
-
- def bkrmdir(self, path):
- if os.path.exists(path):
- sta = os.stat(path)
- oldmode = stat.S_IMODE(sta.st_mode)
- self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
- self._save(_RMDIR, path)
- __built_in_rmdir__(path)
-
- def bkmkdir(self, path, mode):
- if not os.path.exists(path):
- self._save(_MKDIR, path)
- __built_in_mkdir__(path, mode)
-
- def bkchmod(self, path, mode):
- if self._tryhook(path) and os.path.exists(path):
- sta = os.stat(path)
- oldmode = stat.S_IMODE(sta.st_mode)
- self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
- __built_in_chmod__(path, mode)
-
- def rollback(self):
- if self.zip:
- self.zip.close()
- self.zip = None
- index = len(self.rlist) - 1
- while index >= 0:
- item = self.rlist[index]
- exist = os.path.exists(item.path)
- if item.action == _MKFILE and exist:
- #if not os.access(item.path, os.W_OK):
- # os.chmod(item.path, S_IWUSR)
- __built_in_remove__(item.path)
- elif item.action == _RMFILE and not exist:
- if not self.zip:
- self.zip = zipfile.ZipFile(self.zipfile, 'r', zipfile.ZIP_DEFLATED)
- arcname = os.path.normpath(item.path)
- arcname = arcname[len(self.workspace)+1:].encode('utf_8')
- if os.sep != "/" and os.sep in arcname:
- arcname = arcname.replace(os.sep, '/')
- mtime = self.zip.getinfo(arcname).date_time
- content = self.zip.read(arcname)
- filep = __built_in_open__(item.path, "wb")
- filep.write(content)
- filep.close()
- intime = time.mktime(mtime + (0, 0, 0))
- os.utime(item.path, (intime, intime))
- elif item.action == _MKDIR and exist:
- while True:
- try:
- __built_in_rmdir__(item.path)
- break
- except IOError:
- # Sleep a short time and try again
- # The anti-virus software may delay the file removal in this directory
- sleep(0.1)
- elif item.action == _RMDIR and not exist:
- __built_in_mkdir__(item.path)
- elif item.action == _CHMOD and exist:
- try:
- __built_in_chmod__(item.path, item.mode)
- except EnvironmentError:
- pass
- index -= 1
- self.commit()
-
- def commit(self):
- if self.zip:
- self.zip.close()
- __built_in_remove__(self.zipfile)
-
- # Check if path needs to be hooked
- def _tryhook(self, path):
- path = os.path.normpath(path)
- works = self.workspace if str(self.workspace).endswith(os.sep) else (self.workspace + os.sep)
- if not path.startswith(works):
- return ''
- for exceptdir in gEXCEPTION_LIST:
- full = os.path.join(self.workspace, exceptdir)
- if full == path or path.startswith(full + os.sep) or os.path.split(full)[0] == path:
- return ''
- return path[len(self.workspace)+1:]
-
-def _hookrm(path):
- if GlobalData.gRECOVERMGR:
- GlobalData.gRECOVERMGR.bkrmfile(path)
- else:
- __built_in_remove__(path)
-
-def _hookmkdir(path, mode=0o777):
- if GlobalData.gRECOVERMGR:
- GlobalData.gRECOVERMGR.bkmkdir(path, mode)
- else:
- __built_in_mkdir__(path, mode)
-
-def _hookrmdir(path):
- if GlobalData.gRECOVERMGR:
- GlobalData.gRECOVERMGR.bkrmdir(path)
- else:
- __built_in_rmdir__(path)
-
-def _hookmkfile(path, mode='r', bufsize=-1):
- if GlobalData.gRECOVERMGR:
- return GlobalData.gRECOVERMGR.bkmkfile(path, mode, bufsize)
- return __built_in_open__(path, mode, bufsize)
-
-def _hookchmod(path, mode):
- if GlobalData.gRECOVERMGR:
- GlobalData.gRECOVERMGR.bkchmod(path, mode)
- else:
- __built_in_chmod__(path, mode)
-
-def SetRecoverMgr(mgr):
- GlobalData.gRECOVERMGR = mgr
-
-os.remove = _hookrm
-os.mkdir = _hookmkdir
-os.rmdir = _hookrmdir
-os.chmod = _hookchmod
-__FileHookOpen__ = _hookmkfile
+## @file
+# This file hooks file and directory creation and removal
+#
+# Copyright (c) 2014 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+#
+
+'''
+File hook
+'''
+
+import os
+import stat
+import time
+import zipfile
+from time import sleep
+from Library import GlobalData
+
+__built_in_remove__ = os.remove
+__built_in_mkdir__ = os.mkdir
+__built_in_rmdir__ = os.rmdir
+__built_in_chmod__ = os.chmod
+__built_in_open__ = open
+
+_RMFILE = 0
+_MKFILE = 1
+_RMDIR = 2
+_MKDIR = 3
+_CHMOD = 4
+
+gBACKUPFILE = 'file.backup'
+gEXCEPTION_LIST = ['Conf'+os.sep+'DistributionPackageDatabase.db', '.tmp', gBACKUPFILE]
+
+class _PathInfo:
+ def __init__(self, action, path, mode=-1):
+ self.action = action
+ self.path = path
+ self.mode = mode
+
+class RecoverMgr:
+ def __init__(self, workspace):
+ self.rlist = []
+ self.zip = None
+ self.workspace = os.path.normpath(workspace)
+ self.backupfile = gBACKUPFILE
+ self.zipfile = os.path.join(self.workspace, gBACKUPFILE)
+
+ def _createzip(self):
+ if self.zip:
+ return
+ self.zip = zipfile.ZipFile(self.zipfile, 'w', zipfile.ZIP_DEFLATED)
+
+ def _save(self, tmp, path):
+ if not self._tryhook(path):
+ return
+ self.rlist.append(_PathInfo(tmp, path))
+
+ def bkrmfile(self, path):
+ arc = self._tryhook(path)
+ if arc and os.path.isfile(path):
+ self._createzip()
+ self.zip.write(path, arc.encode('utf_8'))
+ sta = os.stat(path)
+ oldmode = stat.S_IMODE(sta.st_mode)
+ self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
+ self.rlist.append(_PathInfo(_RMFILE, path))
+ __built_in_remove__(path)
+
+ def bkmkfile(self, path, mode, bufsize):
+ if not os.path.exists(path):
+ self._save(_MKFILE, path)
+ return __built_in_open__(path, mode, bufsize)
+
+ def bkrmdir(self, path):
+ if os.path.exists(path):
+ sta = os.stat(path)
+ oldmode = stat.S_IMODE(sta.st_mode)
+ self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
+ self._save(_RMDIR, path)
+ __built_in_rmdir__(path)
+
+ def bkmkdir(self, path, mode):
+ if not os.path.exists(path):
+ self._save(_MKDIR, path)
+ __built_in_mkdir__(path, mode)
+
+ def bkchmod(self, path, mode):
+ if self._tryhook(path) and os.path.exists(path):
+ sta = os.stat(path)
+ oldmode = stat.S_IMODE(sta.st_mode)
+ self.rlist.append(_PathInfo(_CHMOD, path, oldmode))
+ __built_in_chmod__(path, mode)
+
+ def rollback(self):
+ if self.zip:
+ self.zip.close()
+ self.zip = None
+ index = len(self.rlist) - 1
+ while index >= 0:
+ item = self.rlist[index]
+ exist = os.path.exists(item.path)
+ if item.action == _MKFILE and exist:
+ #if not os.access(item.path, os.W_OK):
+ # os.chmod(item.path, S_IWUSR)
+ __built_in_remove__(item.path)
+ elif item.action == _RMFILE and not exist:
+ if not self.zip:
+ self.zip = zipfile.ZipFile(self.zipfile, 'r', zipfile.ZIP_DEFLATED)
+ arcname = os.path.normpath(item.path)
+ arcname = arcname[len(self.workspace)+1:].encode('utf_8')
+ if os.sep != "/" and os.sep in arcname:
+ arcname = arcname.replace(os.sep, '/')
+ mtime = self.zip.getinfo(arcname).date_time
+ content = self.zip.read(arcname)
+ filep = __built_in_open__(item.path, "wb")
+ filep.write(content)
+ filep.close()
+ intime = time.mktime(mtime + (0, 0, 0))
+ os.utime(item.path, (intime, intime))
+ elif item.action == _MKDIR and exist:
+ while True:
+ try:
+ __built_in_rmdir__(item.path)
+ break
+ except IOError:
+ # Sleep a short time and try again
+ # The anti-virus software may delay the file removal in this directory
+ sleep(0.1)
+ elif item.action == _RMDIR and not exist:
+ __built_in_mkdir__(item.path)
+ elif item.action == _CHMOD and exist:
+ try:
+ __built_in_chmod__(item.path, item.mode)
+ except EnvironmentError:
+ pass
+ index -= 1
+ self.commit()
+
+ def commit(self):
+ if self.zip:
+ self.zip.close()
+ __built_in_remove__(self.zipfile)
+
+ # Check if path needs to be hooked
+ def _tryhook(self, path):
+ path = os.path.normpath(path)
+ works = self.workspace if str(self.workspace).endswith(os.sep) else (self.workspace + os.sep)
+ if not path.startswith(works):
+ return ''
+ for exceptdir in gEXCEPTION_LIST:
+ full = os.path.join(self.workspace, exceptdir)
+ if full == path or path.startswith(full + os.sep) or os.path.split(full)[0] == path:
+ return ''
+ return path[len(self.workspace)+1:]
+
+def _hookrm(path):
+ if GlobalData.gRECOVERMGR:
+ GlobalData.gRECOVERMGR.bkrmfile(path)
+ else:
+ __built_in_remove__(path)
+
+def _hookmkdir(path, mode=0o777):
+ if GlobalData.gRECOVERMGR:
+ GlobalData.gRECOVERMGR.bkmkdir(path, mode)
+ else:
+ __built_in_mkdir__(path, mode)
+
+def _hookrmdir(path):
+ if GlobalData.gRECOVERMGR:
+ GlobalData.gRECOVERMGR.bkrmdir(path)
+ else:
+ __built_in_rmdir__(path)
+
+def _hookmkfile(path, mode='r', bufsize=-1):
+ if GlobalData.gRECOVERMGR:
+ return GlobalData.gRECOVERMGR.bkmkfile(path, mode, bufsize)
+ return __built_in_open__(path, mode, bufsize)
+
+def _hookchmod(path, mode):
+ if GlobalData.gRECOVERMGR:
+ GlobalData.gRECOVERMGR.bkchmod(path, mode)
+ else:
+ __built_in_chmod__(path, mode)
+
+def SetRecoverMgr(mgr):
+ GlobalData.gRECOVERMGR = mgr
+
+os.remove = _hookrm
+os.mkdir = _hookmkdir
+os.rmdir = _hookrmdir
+os.chmod = _hookchmod
+__FileHookOpen__ = _hookmkfile
diff --git a/BaseTools/Source/Python/UPT/Core/IpiDb.py b/BaseTools/Source/Python/UPT/Core/IpiDb.py
index 69895e11cd..7699a9c54f 100644
--- a/BaseTools/Source/Python/UPT/Core/IpiDb.py
+++ b/BaseTools/Source/Python/UPT/Core/IpiDb.py
@@ -1,922 +1,922 @@
-## @file
-# This file is for installed package information database operations
-#
-# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-#
-
-'''
-IpiDb
-'''
-
-##
-# Import Modules
-#
-import sqlite3
-import os.path
-import time
-
-import Logger.Log as Logger
-from Logger import StringTable as ST
-from Logger.ToolError import UPT_ALREADY_RUNNING_ERROR
-from Logger.ToolError import UPT_DB_UPDATE_ERROR
-import platform as pf
-
-## IpiDb
-#
-# This class represents the installed package information database
-# Add/Remove/Get installed distribution package information here.
-#
-#
-# @param object: Inherited from object class
-# @param DbPath: A string for the path of the database
-#
-#
-class IpiDatabase(object):
- def __init__(self, DbPath, Workspace):
- Dir = os.path.dirname(DbPath)
- if not os.path.isdir(Dir):
- os.mkdir(Dir)
- self.Conn = sqlite3.connect(u''.join(DbPath), isolation_level='DEFERRED')
- self.Conn.execute("PRAGMA page_size=4096")
- self.Conn.execute("PRAGMA synchronous=OFF")
- self.Cur = self.Conn.cursor()
- self.DpTable = 'DpInfo'
- self.PkgTable = 'PkgInfo'
- self.ModInPkgTable = 'ModInPkgInfo'
- self.StandaloneModTable = 'StandaloneModInfo'
- self.ModDepexTable = 'ModDepexInfo'
- self.DpFileListTable = 'DpFileListInfo'
- self.DummyTable = 'Dummy'
- self.Workspace = os.path.normpath(Workspace)
-
- ## Initialize build database
- #
- #
- def InitDatabase(self, SkipLock = False):
- Logger.Verbose(ST.MSG_INIT_IPI_START)
- if not SkipLock:
- try:
- #
- # Create a dummy table, if already existed,
- # then UPT is already running
- #
- SqlCommand = """
- create table %s (
- Dummy TEXT NOT NULL,
- PRIMARY KEY (Dummy)
- )""" % self.DummyTable
- self.Cur.execute(SqlCommand)
- self.Conn.commit()
- except sqlite3.OperationalError:
- Logger.Error("UPT",
- UPT_ALREADY_RUNNING_ERROR,
- ST.ERR_UPT_ALREADY_RUNNING_ERROR
- )
-
- #
- # Create new table
- #
- SqlCommand = """
- create table IF NOT EXISTS %s (
- DpGuid TEXT NOT NULL,DpVersion TEXT NOT NULL,
- InstallTime REAL NOT NULL,
- NewPkgFileName TEXT NOT NULL,
- PkgFileName TEXT NOT NULL,
- RePackage TEXT NOT NULL,
- PRIMARY KEY (DpGuid, DpVersion)
- )""" % self.DpTable
- self.Cur.execute(SqlCommand)
-
- SqlCommand = """
- create table IF NOT EXISTS %s (
- FilePath TEXT NOT NULL,
- DpGuid TEXT,
- DpVersion TEXT,
- Md5Sum TEXT,
- PRIMARY KEY (FilePath)
- )""" % self.DpFileListTable
- self.Cur.execute(SqlCommand)
-
- SqlCommand = """
- create table IF NOT EXISTS %s (
- PackageGuid TEXT NOT NULL,
- PackageVersion TEXT NOT NULL,
- InstallTime REAL NOT NULL,
- DpGuid TEXT,
- DpVersion TEXT,
- InstallPath TEXT NOT NULL,
- PRIMARY KEY (PackageGuid, PackageVersion, InstallPath)
- )""" % self.PkgTable
- self.Cur.execute(SqlCommand)
-
- SqlCommand = """
- create table IF NOT EXISTS %s (
- ModuleGuid TEXT NOT NULL,
- ModuleVersion TEXT NOT NULL,
- ModuleName TEXT NOT NULL,
- InstallTime REAL NOT NULL,
- PackageGuid TEXT,
- PackageVersion TEXT,
- InstallPath TEXT NOT NULL,
- PRIMARY KEY (ModuleGuid, ModuleVersion, ModuleName, InstallPath)
- )""" % self.ModInPkgTable
- self.Cur.execute(SqlCommand)
-
- SqlCommand = """
- create table IF NOT EXISTS %s (
- ModuleGuid TEXT NOT NULL,
- ModuleVersion TEXT NOT NULL,
- ModuleName TEXT NOT NULL,
- InstallTime REAL NOT NULL,
- DpGuid TEXT,
- DpVersion TEXT,
- InstallPath TEXT NOT NULL,
- PRIMARY KEY (ModuleGuid, ModuleVersion, ModuleName, InstallPath)
- )""" % self.StandaloneModTable
- self.Cur.execute(SqlCommand)
-
- SqlCommand = """
- create table IF NOT EXISTS %s (
- ModuleGuid TEXT NOT NULL,
- ModuleVersion TEXT NOT NULL,
- ModuleName TEXT NOT NULL,
- InstallPath TEXT NOT NULL,
- DepexGuid TEXT,
- DepexVersion TEXT
- )""" % self.ModDepexTable
- self.Cur.execute(SqlCommand)
-
- self.Conn.commit()
-
- Logger.Verbose(ST.MSG_INIT_IPI_FINISH)
-
- def RollBack(self):
- self.Conn.rollback()
-
- def Commit(self):
- self.Conn.commit()
-
- ## Add a distribution install information from DpObj
- #
- # @param DpObj:
- # @param NewDpPkgFileName: New DpPkg File Name
- # @param DpPkgFileName: DpPkg File Name
- # @param RePackage: A RePackage
- #
- def AddDPObject(self, DpObj, NewDpPkgFileName, DpPkgFileName, RePackage):
- try:
- for PkgKey in DpObj.PackageSurfaceArea.keys():
- PkgGuid = PkgKey[0]
- PkgVersion = PkgKey[1]
- PkgInstallPath = PkgKey[2]
- self._AddPackage(PkgGuid, PkgVersion, DpObj.Header.GetGuid(), \
- DpObj.Header.GetVersion(), PkgInstallPath)
- PkgObj = DpObj.PackageSurfaceArea[PkgKey]
- for ModKey in PkgObj.GetModuleDict().keys():
- ModGuid = ModKey[0]
- ModVersion = ModKey[1]
- ModName = ModKey[2]
- ModInstallPath = ModKey[3]
- ModInstallPath = \
- os.path.normpath(os.path.join(PkgInstallPath, ModInstallPath))
- self._AddModuleInPackage(ModGuid, ModVersion, ModName, PkgGuid, \
- PkgVersion, ModInstallPath)
- ModObj = PkgObj.GetModuleDict()[ModKey]
- for Dep in ModObj.GetPackageDependencyList():
- DepexGuid = Dep.GetGuid()
- DepexVersion = Dep.GetVersion()
- self._AddModuleDepex(ModGuid, ModVersion, ModName, ModInstallPath, \
- DepexGuid, DepexVersion)
- for (FilePath, Md5Sum) in PkgObj.FileList:
- self._AddDpFilePathList(DpObj.Header.GetGuid(), \
- DpObj.Header.GetVersion(), FilePath, \
- Md5Sum)
-
- for ModKey in DpObj.ModuleSurfaceArea.keys():
- ModGuid = ModKey[0]
- ModVersion = ModKey[1]
- ModName = ModKey[2]
- ModInstallPath = ModKey[3]
- self._AddStandaloneModule(ModGuid, ModVersion, ModName, \
- DpObj.Header.GetGuid(), \
- DpObj.Header.GetVersion(), \
- ModInstallPath)
- ModObj = DpObj.ModuleSurfaceArea[ModKey]
- for Dep in ModObj.GetPackageDependencyList():
- DepexGuid = Dep.GetGuid()
- DepexVersion = Dep.GetVersion()
- self._AddModuleDepex(ModGuid, ModVersion, ModName, ModInstallPath, \
- DepexGuid, DepexVersion)
- for (Path, Md5Sum) in ModObj.FileList:
- self._AddDpFilePathList(DpObj.Header.GetGuid(), \
- DpObj.Header.GetVersion(), \
- Path, Md5Sum)
-
- #
- # add tool/misc files
- #
- for (Path, Md5Sum) in DpObj.FileList:
- self._AddDpFilePathList(DpObj.Header.GetGuid(), \
- DpObj.Header.GetVersion(), Path, Md5Sum)
-
- self._AddDp(DpObj.Header.GetGuid(), DpObj.Header.GetVersion(), \
- NewDpPkgFileName, DpPkgFileName, RePackage)
-
- except sqlite3.IntegrityError as DetailMsg:
- Logger.Error("UPT",
- UPT_DB_UPDATE_ERROR,
- ST.ERR_UPT_DB_UPDATE_ERROR,
- ExtraData = DetailMsg
- )
-
- ## Add a distribution install information
- #
- # @param Guid Guid of the distribution package
- # @param Version Version of the distribution package
- # @param NewDpFileName the saved filename of distribution package file
- # @param DistributionFileName the filename of distribution package file
- #
- def _AddDp(self, Guid, Version, NewDpFileName, DistributionFileName, \
- RePackage):
-
- if Version is None or len(Version.strip()) == 0:
- Version = 'N/A'
-
- #
- # Add newly installed DP information to DB.
- #
- if NewDpFileName is None or len(NewDpFileName.strip()) == 0:
- PkgFileName = 'N/A'
- else:
- PkgFileName = NewDpFileName
- CurrentTime = time.time()
- SqlCommand = \
- """insert into %s values('%s', '%s', %s, '%s', '%s', '%s')""" % \
- (self.DpTable, Guid, Version, CurrentTime, PkgFileName, \
- DistributionFileName, str(RePackage).upper())
- self.Cur.execute(SqlCommand)
-
-
- ## Add a file list from DP
- #
- # @param DpGuid: A DpGuid
- # @param DpVersion: A DpVersion
- # @param Path: A Path
- # @param Path: A Md5Sum
- #
- def _AddDpFilePathList(self, DpGuid, DpVersion, Path, Md5Sum):
- Path = os.path.normpath(Path)
- if pf.system() == 'Windows':
- if Path.startswith(self.Workspace):
- Path = Path[len(self.Workspace):]
- else:
- if Path.startswith(self.Workspace + os.sep):
- Path = Path[len(self.Workspace)+1:]
- SqlCommand = """insert into %s values('%s', '%s', '%s', '%s')""" % \
- (self.DpFileListTable, Path, DpGuid, DpVersion, Md5Sum)
-
- self.Cur.execute(SqlCommand)
-
- ## Add a package install information
- #
- # @param Guid: A package guid
- # @param Version: A package version
- # @param DpGuid: A DpGuid
- # @param DpVersion: A DpVersion
- # @param Path: A Path
- #
- def _AddPackage(self, Guid, Version, DpGuid=None, DpVersion=None, Path=''):
-
- if Version is None or len(Version.strip()) == 0:
- Version = 'N/A'
-
- if DpGuid is None or len(DpGuid.strip()) == 0:
- DpGuid = 'N/A'
-
- if DpVersion is None or len(DpVersion.strip()) == 0:
- DpVersion = 'N/A'
-
- #
- # Add newly installed package information to DB.
- #
- CurrentTime = time.time()
- SqlCommand = \
- """insert into %s values('%s', '%s', %s, '%s', '%s', '%s')""" % \
- (self.PkgTable, Guid, Version, CurrentTime, DpGuid, DpVersion, Path)
- self.Cur.execute(SqlCommand)
-
- ## Add a module that from a package install information
- #
- # @param Guid: Module Guid
- # @param Version: Module version
- # @param Name: Module Name
- # @param PkgGuid: Package Guid
- # @param PkgVersion: Package version
- # @param Path: Package relative path that module installs
- #
- def _AddModuleInPackage(self, Guid, Version, Name, PkgGuid=None, \
- PkgVersion=None, Path=''):
-
- if Version is None or len(Version.strip()) == 0:
- Version = 'N/A'
-
- if PkgGuid is None or len(PkgGuid.strip()) == 0:
- PkgGuid = 'N/A'
-
- if PkgVersion is None or len(PkgVersion.strip()) == 0:
- PkgVersion = 'N/A'
-
- if os.name == 'posix':
- Path = Path.replace('\\', os.sep)
- else:
- Path = Path.replace('/', os.sep)
-
- #
- # Add module from package information to DB.
- #
- CurrentTime = time.time()
- SqlCommand = \
- """insert into %s values('%s', '%s', '%s', %s, '%s', '%s', '%s')""" % \
- (self.ModInPkgTable, Guid, Version, Name, CurrentTime, PkgGuid, PkgVersion, \
- Path)
- self.Cur.execute(SqlCommand)
-
- ## Add a module that is standalone install information
- #
- # @param Guid: a module Guid
- # @param Version: a module Version
- # @param Name: a module name
- # @param DpGuid: a DpGuid
- # @param DpVersion: a DpVersion
- # @param Path: path
- #
- def _AddStandaloneModule(self, Guid, Version, Name, DpGuid=None, \
- DpVersion=None, Path=''):
-
- if Version is None or len(Version.strip()) == 0:
- Version = 'N/A'
-
- if DpGuid is None or len(DpGuid.strip()) == 0:
- DpGuid = 'N/A'
-
- if DpVersion is None or len(DpVersion.strip()) == 0:
- DpVersion = 'N/A'
-
- #
- # Add module standalone information to DB.
- #
- CurrentTime = time.time()
- SqlCommand = \
- """insert into %s values('%s', '%s', '%s', %s, '%s', '%s', '%s')""" % \
- (self.StandaloneModTable, Guid, Version, Name, CurrentTime, DpGuid, \
- DpVersion, Path)
- self.Cur.execute(SqlCommand)
-
- ## Add a module depex
- #
- # @param Guid: a module Guid
- # @param Version: a module Version
- # @param Name: a module name
- # @param DepexGuid: a module DepexGuid
- # @param DepexVersion: a module DepexVersion
- #
- def _AddModuleDepex(self, Guid, Version, Name, Path, DepexGuid=None, \
- DepexVersion=None):
-
- if DepexGuid is None or len(DepexGuid.strip()) == 0:
- DepexGuid = 'N/A'
-
- if DepexVersion is None or len(DepexVersion.strip()) == 0:
- DepexVersion = 'N/A'
-
- if os.name == 'posix':
- Path = Path.replace('\\', os.sep)
- else:
- Path = Path.replace('/', os.sep)
-
- #
- # Add module depex information to DB.
- #
- SqlCommand = """insert into %s values('%s', '%s', '%s', '%s', '%s', '%s')"""\
- % (self.ModDepexTable, Guid, Version, Name, Path, DepexGuid, DepexVersion)
- self.Cur.execute(SqlCommand)
-
- ## Remove a distribution install information, if no version specified,
- # remove all DPs with this Guid.
- #
- # @param DpGuid: guid of dpex
- # @param DpVersion: version of dpex
- #
- def RemoveDpObj(self, DpGuid, DpVersion):
-
- PkgList = self.GetPackageListFromDp(DpGuid, DpVersion)
- #
- # delete from ModDepex the standalone module's dependency
- #
- SqlCommand = \
- """delete from ModDepexInfo where ModDepexInfo.ModuleGuid in
- (select ModuleGuid from StandaloneModInfo as B where B.DpGuid = '%s'
- and B.DpVersion = '%s')
- and ModDepexInfo.ModuleVersion in
- (select ModuleVersion from StandaloneModInfo as B
- where B.DpGuid = '%s' and B.DpVersion = '%s')
- and ModDepexInfo.ModuleName in
- (select ModuleName from StandaloneModInfo as B
- where B.DpGuid = '%s' and B.DpVersion = '%s')
- and ModDepexInfo.InstallPath in
- (select InstallPath from StandaloneModInfo as B
- where B.DpGuid = '%s' and B.DpVersion = '%s') """ % \
- (DpGuid, DpVersion, DpGuid, DpVersion, DpGuid, DpVersion, DpGuid, DpVersion)
-
- self.Cur.execute(SqlCommand)
- #
- # delete from ModDepex the from pkg module's dependency
- #
- for Pkg in PkgList:
-
- SqlCommand = \
- """delete from ModDepexInfo where ModDepexInfo.ModuleGuid in
- (select ModuleGuid from ModInPkgInfo
- where ModInPkgInfo.PackageGuid ='%s' and
- ModInPkgInfo.PackageVersion = '%s')
- and ModDepexInfo.ModuleVersion in
- (select ModuleVersion from ModInPkgInfo
- where ModInPkgInfo.PackageGuid ='%s' and
- ModInPkgInfo.PackageVersion = '%s')
- and ModDepexInfo.ModuleName in
- (select ModuleName from ModInPkgInfo
- where ModInPkgInfo.PackageGuid ='%s' and
- ModInPkgInfo.PackageVersion = '%s')
- and ModDepexInfo.InstallPath in
- (select InstallPath from ModInPkgInfo where
- ModInPkgInfo.PackageGuid ='%s'
- and ModInPkgInfo.PackageVersion = '%s')""" \
- % (Pkg[0], Pkg[1], Pkg[0], Pkg[1], Pkg[0], Pkg[1], Pkg[0], Pkg[1])
-
- self.Cur.execute(SqlCommand)
- #
- # delete the standalone module
- #
- SqlCommand = \
- """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.StandaloneModTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
- #
- # delete the from pkg module
- #
- for Pkg in PkgList:
- SqlCommand = \
- """delete from %s where %s.PackageGuid ='%s'
- and %s.PackageVersion = '%s'""" % \
- (self.ModInPkgTable, self.ModInPkgTable, Pkg[0], \
- self.ModInPkgTable, Pkg[1])
- self.Cur.execute(SqlCommand)
- #
- # delete packages
- #
- SqlCommand = \
- """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.PkgTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
- #
- # delete file list from DP
- #
- SqlCommand = \
- """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.DpFileListTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
- #
- # delete DP
- #
- SqlCommand = \
- """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.DpTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- #self.Conn.commit()
-
- ## Get a list of distribution install information.
- #
- # @param Guid: distribution package guid
- # @param Version: distribution package version
- #
- def GetDp(self, Guid, Version):
-
- if Version is None or len(Version.strip()) == 0:
- Version = 'N/A'
- Logger.Verbose(ST.MSG_GET_DP_INSTALL_LIST)
- (DpGuid, DpVersion) = (Guid, Version)
- SqlCommand = """select * from %s where DpGuid ='%s'""" % \
- (self.DpTable, DpGuid)
- self.Cur.execute(SqlCommand)
-
- else:
- Logger.Verbose(ST.MSG_GET_DP_INSTALL_INFO_START)
- (DpGuid, DpVersion) = (Guid, Version)
- SqlCommand = \
- """select * from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.DpTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- DpList = []
- for DpInfo in self.Cur:
- DpGuid = DpInfo[0]
- DpVersion = DpInfo[1]
- InstallTime = DpInfo[2]
- PkgFileName = DpInfo[3]
- DpList.append((DpGuid, DpVersion, InstallTime, PkgFileName))
-
- Logger.Verbose(ST.MSG_GET_DP_INSTALL_INFO_FINISH)
- return DpList
-
- ## Get a list of distribution install dirs
- #
- # @param Guid: distribution package guid
- # @param Version: distribution package version
- #
- def GetDpInstallDirList(self, Guid, Version):
- SqlCommand = """select InstallPath from PkgInfo where DpGuid = '%s' and DpVersion = '%s'""" % (Guid, Version)
- self.Cur.execute(SqlCommand)
- DirList = []
- for Result in self.Cur:
- if Result[0] not in DirList:
- DirList.append(Result[0])
-
- SqlCommand = """select InstallPath from StandaloneModInfo where DpGuid = '%s' and DpVersion = '%s'""" % \
- (Guid, Version)
- self.Cur.execute(SqlCommand)
- for Result in self.Cur:
- if Result[0] not in DirList:
- DirList.append(Result[0])
-
- return DirList
-
-
- ## Get a list of distribution install file path information.
- #
- # @param Guid: distribution package guid
- # @param Version: distribution package version
- #
- def GetDpFileList(self, Guid, Version):
-
- (DpGuid, DpVersion) = (Guid, Version)
- SqlCommand = \
- """select * from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
- (self.DpFileListTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- PathList = []
- for Result in self.Cur:
- Path = Result[0]
- Md5Sum = Result[3]
- PathList.append((os.path.join(self.Workspace, Path), Md5Sum))
-
- return PathList
-
- ## Get files' repackage attribute if present that are installed into current workspace
- #
- # @retval FileDict: a Dict of file, key is file path, value is (DpGuid, DpVersion, NewDpFileName, RePackage)
- #
- def GetRePkgDict(self):
- SqlCommand = """select * from %s """ % (self.DpTable)
- self.Cur.execute(SqlCommand)
-
- DpInfoList = []
- for Result in self.Cur:
- DpInfoList.append(Result)
-
- FileDict = {}
- for Result in DpInfoList:
- DpGuid = Result[0]
- DpVersion = Result[1]
- NewDpFileName = Result[3]
- RePackage = Result[5]
- if RePackage == 'TRUE':
- RePackage = True
- else:
- RePackage = False
- for FileInfo in self.GetDpFileList(DpGuid, DpVersion):
- PathInfo = FileInfo[0]
- FileDict[PathInfo] = DpGuid, DpVersion, NewDpFileName, RePackage
-
- return FileDict
-
- ## Get (Guid, Version) from distribution file name information.
- #
- # @param DistributionFile: Distribution File
- #
- def GetDpByName(self, DistributionFile):
- SqlCommand = """select * from %s where NewPkgFileName = '%s'""" % \
- (self.DpTable, DistributionFile)
- self.Cur.execute(SqlCommand)
-
- for Result in self.Cur:
- DpGuid = Result[0]
- DpVersion = Result[1]
- NewDpFileName = Result[3]
-
- return (DpGuid, DpVersion, NewDpFileName)
- else:
- return (None, None, None)
-
- ## Get a list of package information.
- #
- # @param Guid: package guid
- # @param Version: package version
- #
- def GetPackage(self, Guid, Version, DpGuid='', DpVersion=''):
-
- if DpVersion == '' or DpGuid == '':
-
- (PackageGuid, PackageVersion) = (Guid, Version)
- SqlCommand = """select * from %s where PackageGuid ='%s'
- and PackageVersion = '%s'""" % (self.PkgTable, PackageGuid, \
- PackageVersion)
- self.Cur.execute(SqlCommand)
-
- elif Version is None or len(Version.strip()) == 0:
-
- SqlCommand = """select * from %s where PackageGuid ='%s'""" % \
- (self.PkgTable, Guid)
- self.Cur.execute(SqlCommand)
- else:
- (PackageGuid, PackageVersion) = (Guid, Version)
- SqlCommand = """select * from %s where PackageGuid ='%s' and
- PackageVersion = '%s'
- and DpGuid = '%s' and DpVersion = '%s'""" % \
- (self.PkgTable, PackageGuid, PackageVersion, \
- DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- PkgList = []
- for PkgInfo in self.Cur:
- PkgGuid = PkgInfo[0]
- PkgVersion = PkgInfo[1]
- InstallTime = PkgInfo[2]
- InstallPath = PkgInfo[5]
- PkgList.append((PkgGuid, PkgVersion, InstallTime, DpGuid, \
- DpVersion, InstallPath))
-
- return PkgList
-
-
- ## Get a list of module in package information.
- #
- # @param Guid: A module guid
- # @param Version: A module version
- #
- def GetModInPackage(self, Guid, Version, Name, Path, PkgGuid='', PkgVersion=''):
- (ModuleGuid, ModuleVersion, ModuleName, InstallPath) = (Guid, Version, Name, Path)
- if PkgVersion == '' or PkgGuid == '':
- SqlCommand = """select * from %s where ModuleGuid ='%s' and
- ModuleVersion = '%s' and InstallPath = '%s'
- and ModuleName = '%s'""" % (self.ModInPkgTable, ModuleGuid, \
- ModuleVersion, InstallPath, ModuleName)
- self.Cur.execute(SqlCommand)
- else:
- SqlCommand = """select * from %s where ModuleGuid ='%s' and
- ModuleVersion = '%s' and InstallPath = '%s'
- and ModuleName = '%s' and PackageGuid ='%s'
- and PackageVersion = '%s'
- """ % (self.ModInPkgTable, ModuleGuid, \
- ModuleVersion, InstallPath, ModuleName, PkgGuid, PkgVersion)
- self.Cur.execute(SqlCommand)
-
- ModList = []
- for ModInfo in self.Cur:
- ModGuid = ModInfo[0]
- ModVersion = ModInfo[1]
- InstallTime = ModInfo[2]
- InstallPath = ModInfo[5]
- ModList.append((ModGuid, ModVersion, InstallTime, PkgGuid, \
- PkgVersion, InstallPath))
-
- return ModList
-
- ## Get a list of module standalone.
- #
- # @param Guid: A module guid
- # @param Version: A module version
- #
- def GetStandaloneModule(self, Guid, Version, Name, Path, DpGuid='', DpVersion=''):
- (ModuleGuid, ModuleVersion, ModuleName, InstallPath) = (Guid, Version, Name, Path)
- if DpGuid == '':
- SqlCommand = """select * from %s where ModuleGuid ='%s' and
- ModuleVersion = '%s' and InstallPath = '%s'
- and ModuleName = '%s'""" % (self.StandaloneModTable, ModuleGuid, \
- ModuleVersion, InstallPath, ModuleName)
- self.Cur.execute(SqlCommand)
-
- else:
- SqlCommand = """select * from %s where ModuleGuid ='%s' and
- ModuleVersion = '%s' and InstallPath = '%s' and ModuleName = '%s' and DpGuid ='%s' and DpVersion = '%s'
- """ % (self.StandaloneModTable, ModuleGuid, \
- ModuleVersion, ModuleName, InstallPath, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- ModList = []
- for ModInfo in self.Cur:
- ModGuid = ModInfo[0]
- ModVersion = ModInfo[1]
- InstallTime = ModInfo[2]
- InstallPath = ModInfo[5]
- ModList.append((ModGuid, ModVersion, InstallTime, DpGuid, \
- DpVersion, InstallPath))
-
- return ModList
-
- ## Get a list of module information that comes from DP.
- #
- # @param DpGuid: A Distribution Guid
- # @param DpVersion: A Distribution version
- #
- def GetSModInsPathListFromDp(self, DpGuid, DpVersion):
-
- PathList = []
- SqlCommand = """select InstallPath from %s where DpGuid ='%s'
- and DpVersion = '%s'
- """ % (self.StandaloneModTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- for Result in self.Cur:
- InstallPath = Result[0]
- PathList.append(InstallPath)
-
- return PathList
-
- ## Get a list of package information.
- #
- # @param DpGuid: A Distribution Guid
- # @param DpVersion: A Distribution version
- #
- def GetPackageListFromDp(self, DpGuid, DpVersion):
-
- SqlCommand = """select * from %s where DpGuid ='%s' and
- DpVersion = '%s' """ % (self.PkgTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
-
- PkgList = []
- for PkgInfo in self.Cur:
- PkgGuid = PkgInfo[0]
- PkgVersion = PkgInfo[1]
- InstallPath = PkgInfo[5]
- PkgList.append((PkgGuid, PkgVersion, InstallPath))
-
- return PkgList
-
- ## Get a list of modules that depends on package information from a DP.
- #
- # @param DpGuid: A Distribution Guid
- # @param DpVersion: A Distribution version
- #
- def GetDpDependentModuleList(self, DpGuid, DpVersion):
-
- ModList = []
- PkgList = self.GetPackageListFromDp(DpGuid, DpVersion)
- if len(PkgList) > 0:
- return ModList
-
- for Pkg in PkgList:
- #
- # get all in-package modules that depends on current
- # Pkg (Guid match, Version match or NA) but not belong to
- # current Pkg
- #
- SqlCommand = """select t1.ModuleGuid, t1.ModuleVersion,
- t1.InstallPath from %s as t1, %s as t2 where
- t1.ModuleGuid = t2.ModuleGuid and
- t1.ModuleVersion = t2.ModuleVersion and t2.DepexGuid ='%s'
- and (t2.DepexVersion = '%s' or t2.DepexVersion = 'N/A') and
- t1.PackageGuid != '%s' and t1.PackageVersion != '%s'
- """ % (self.ModInPkgTable, \
- self.ModDepexTable, Pkg[0], Pkg[1], Pkg[0], \
- Pkg[1])
- self.Cur.execute(SqlCommand)
- for ModInfo in self.Cur:
- ModGuid = ModInfo[0]
- ModVersion = ModInfo[1]
- InstallPath = ModInfo[2]
- ModList.append((ModGuid, ModVersion, InstallPath))
-
- #
- # get all modules from standalone modules that depends on current
- #Pkg (Guid match, Version match or NA) but not in current dp
- #
- SqlCommand = \
- """select t1.ModuleGuid, t1.ModuleVersion, t1.InstallPath
- from %s as t1, %s as t2 where t1.ModuleGuid = t2.ModuleGuid and
- t1.ModuleVersion = t2.ModuleVersion and t2.DepexGuid ='%s'
- and (t2.DepexVersion = '%s' or t2.DepexVersion = 'N/A') and
- t1.DpGuid != '%s' and t1.DpVersion != '%s'
- """ % \
- (self.StandaloneModTable, self.ModDepexTable, Pkg[0], \
- Pkg[1], DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
- for ModInfo in self.Cur:
- ModGuid = ModInfo[0]
- ModVersion = ModInfo[1]
- InstallPath = ModInfo[2]
- ModList.append((ModGuid, ModVersion, InstallPath))
-
-
- return ModList
-
- ## Get Dp's list of modules.
- #
- # @param DpGuid: A Distribution Guid
- # @param DpVersion: A Distribution version
- #
- def GetDpModuleList(self, DpGuid, DpVersion):
- ModList = []
- #
- # get Dp module list from the DpFileList table
- #
- SqlCommand = """select FilePath
- from %s
- where DpGuid = '%s' and DpVersion = '%s' and
- FilePath like '%%.inf'
- """ % (self.DpFileListTable, DpGuid, DpVersion)
- self.Cur.execute(SqlCommand)
- for ModuleInfo in self.Cur:
- FilePath = ModuleInfo[0]
- ModList.append(os.path.join(self.Workspace, FilePath))
-
- return ModList
-
-
- ## Get a module depex
- #
- # @param DpGuid: A module Guid
- # @param DpVersion: A module version
- # @param Path:
- #
- def GetModuleDepex(self, Guid, Version, Path):
-
- #
- # Get module depex information to DB.
- #
- SqlCommand = """select * from %s where ModuleGuid ='%s' and
- ModuleVersion = '%s' and InstallPath ='%s'
- """ % (self.ModDepexTable, Guid, Version, Path)
- self.Cur.execute(SqlCommand)
-
-
- DepexList = []
- for DepInfo in self.Cur:
- DepexGuid = DepInfo[3]
- DepexVersion = DepInfo[4]
- DepexList.append((DepexGuid, DepexVersion))
-
- return DepexList
-
- ## Inventory the distribution installed to current workspace
- #
- # Inventory the distribution installed to current workspace
- #
- def InventoryDistInstalled(self):
- SqlCommand = """select * from %s """ % (self.DpTable)
- self.Cur.execute(SqlCommand)
-
- DpInfoList = []
- for Result in self.Cur:
- DpGuid = Result[0]
- DpVersion = Result[1]
- DpAliasName = Result[3]
- DpFileName = Result[4]
- DpInfoList.append((DpGuid, DpVersion, DpFileName, DpAliasName))
-
- return DpInfoList
-
- ## Close entire database
- #
- # Close the connection and cursor
- #
- def CloseDb(self):
- #
- # drop the dummy table
- #
- SqlCommand = """
- drop table IF EXISTS %s
- """ % self.DummyTable
- self.Cur.execute(SqlCommand)
- self.Conn.commit()
-
- self.Cur.close()
- self.Conn.close()
-
- ## Convert To Sql String
- #
- # 1. Replace "'" with "''" in each item of StringList
- #
- # @param StringList: A list for strings to be converted
- #
- def __ConvertToSqlString(self, StringList):
- if self.DpTable:
- pass
- return list(map(lambda s: s.replace("'", "''"), StringList))
-
-
-
-
+## @file
+# This file is for installed package information database operations
+#
+# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+#
+
+'''
+IpiDb
+'''
+
+##
+# Import Modules
+#
+import sqlite3
+import os.path
+import time
+
+import Logger.Log as Logger
+from Logger import StringTable as ST
+from Logger.ToolError import UPT_ALREADY_RUNNING_ERROR
+from Logger.ToolError import UPT_DB_UPDATE_ERROR
+import platform as pf
+
+## IpiDb
+#
+# This class represents the installed package information database
+# Add/Remove/Get installed distribution package information here.
+#
+#
+# @param object: Inherited from object class
+# @param DbPath: A string for the path of the database
+#
+#
+class IpiDatabase(object):
+ def __init__(self, DbPath, Workspace):
+ Dir = os.path.dirname(DbPath)
+ if not os.path.isdir(Dir):
+ os.mkdir(Dir)
+ self.Conn = sqlite3.connect(u''.join(DbPath), isolation_level='DEFERRED')
+ self.Conn.execute("PRAGMA page_size=4096")
+ self.Conn.execute("PRAGMA synchronous=OFF")
+ self.Cur = self.Conn.cursor()
+ self.DpTable = 'DpInfo'
+ self.PkgTable = 'PkgInfo'
+ self.ModInPkgTable = 'ModInPkgInfo'
+ self.StandaloneModTable = 'StandaloneModInfo'
+ self.ModDepexTable = 'ModDepexInfo'
+ self.DpFileListTable = 'DpFileListInfo'
+ self.DummyTable = 'Dummy'
+ self.Workspace = os.path.normpath(Workspace)
+
+ ## Initialize build database
+ #
+ #
+ def InitDatabase(self, SkipLock = False):
+ Logger.Verbose(ST.MSG_INIT_IPI_START)
+ if not SkipLock:
+ try:
+ #
+ # Create a dummy table, if already existed,
+ # then UPT is already running
+ #
+ SqlCommand = """
+ create table %s (
+ Dummy TEXT NOT NULL,
+ PRIMARY KEY (Dummy)
+ )""" % self.DummyTable
+ self.Cur.execute(SqlCommand)
+ self.Conn.commit()
+ except sqlite3.OperationalError:
+ Logger.Error("UPT",
+ UPT_ALREADY_RUNNING_ERROR,
+ ST.ERR_UPT_ALREADY_RUNNING_ERROR
+ )
+
+ #
+ # Create new table
+ #
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ DpGuid TEXT NOT NULL,DpVersion TEXT NOT NULL,
+ InstallTime REAL NOT NULL,
+ NewPkgFileName TEXT NOT NULL,
+ PkgFileName TEXT NOT NULL,
+ RePackage TEXT NOT NULL,
+ PRIMARY KEY (DpGuid, DpVersion)
+ )""" % self.DpTable
+ self.Cur.execute(SqlCommand)
+
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ FilePath TEXT NOT NULL,
+ DpGuid TEXT,
+ DpVersion TEXT,
+ Md5Sum TEXT,
+ PRIMARY KEY (FilePath)
+ )""" % self.DpFileListTable
+ self.Cur.execute(SqlCommand)
+
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ PackageGuid TEXT NOT NULL,
+ PackageVersion TEXT NOT NULL,
+ InstallTime REAL NOT NULL,
+ DpGuid TEXT,
+ DpVersion TEXT,
+ InstallPath TEXT NOT NULL,
+ PRIMARY KEY (PackageGuid, PackageVersion, InstallPath)
+ )""" % self.PkgTable
+ self.Cur.execute(SqlCommand)
+
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ ModuleGuid TEXT NOT NULL,
+ ModuleVersion TEXT NOT NULL,
+ ModuleName TEXT NOT NULL,
+ InstallTime REAL NOT NULL,
+ PackageGuid TEXT,
+ PackageVersion TEXT,
+ InstallPath TEXT NOT NULL,
+ PRIMARY KEY (ModuleGuid, ModuleVersion, ModuleName, InstallPath)
+ )""" % self.ModInPkgTable
+ self.Cur.execute(SqlCommand)
+
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ ModuleGuid TEXT NOT NULL,
+ ModuleVersion TEXT NOT NULL,
+ ModuleName TEXT NOT NULL,
+ InstallTime REAL NOT NULL,
+ DpGuid TEXT,
+ DpVersion TEXT,
+ InstallPath TEXT NOT NULL,
+ PRIMARY KEY (ModuleGuid, ModuleVersion, ModuleName, InstallPath)
+ )""" % self.StandaloneModTable
+ self.Cur.execute(SqlCommand)
+
+ SqlCommand = """
+ create table IF NOT EXISTS %s (
+ ModuleGuid TEXT NOT NULL,
+ ModuleVersion TEXT NOT NULL,
+ ModuleName TEXT NOT NULL,
+ InstallPath TEXT NOT NULL,
+ DepexGuid TEXT,
+ DepexVersion TEXT
+ )""" % self.ModDepexTable
+ self.Cur.execute(SqlCommand)
+
+ self.Conn.commit()
+
+ Logger.Verbose(ST.MSG_INIT_IPI_FINISH)
+
+ def RollBack(self):
+ self.Conn.rollback()
+
+ def Commit(self):
+ self.Conn.commit()
+
+ ## Add a distribution install information from DpObj
+ #
+ # @param DpObj:
+ # @param NewDpPkgFileName: New DpPkg File Name
+ # @param DpPkgFileName: DpPkg File Name
+ # @param RePackage: A RePackage
+ #
+ def AddDPObject(self, DpObj, NewDpPkgFileName, DpPkgFileName, RePackage):
+ try:
+ for PkgKey in DpObj.PackageSurfaceArea.keys():
+ PkgGuid = PkgKey[0]
+ PkgVersion = PkgKey[1]
+ PkgInstallPath = PkgKey[2]
+ self._AddPackage(PkgGuid, PkgVersion, DpObj.Header.GetGuid(), \
+ DpObj.Header.GetVersion(), PkgInstallPath)
+ PkgObj = DpObj.PackageSurfaceArea[PkgKey]
+ for ModKey in PkgObj.GetModuleDict().keys():
+ ModGuid = ModKey[0]
+ ModVersion = ModKey[1]
+ ModName = ModKey[2]
+ ModInstallPath = ModKey[3]
+ ModInstallPath = \
+ os.path.normpath(os.path.join(PkgInstallPath, ModInstallPath))
+ self._AddModuleInPackage(ModGuid, ModVersion, ModName, PkgGuid, \
+ PkgVersion, ModInstallPath)
+ ModObj = PkgObj.GetModuleDict()[ModKey]
+ for Dep in ModObj.GetPackageDependencyList():
+ DepexGuid = Dep.GetGuid()
+ DepexVersion = Dep.GetVersion()
+ self._AddModuleDepex(ModGuid, ModVersion, ModName, ModInstallPath, \
+ DepexGuid, DepexVersion)
+ for (FilePath, Md5Sum) in PkgObj.FileList:
+ self._AddDpFilePathList(DpObj.Header.GetGuid(), \
+ DpObj.Header.GetVersion(), FilePath, \
+ Md5Sum)
+
+ for ModKey in DpObj.ModuleSurfaceArea.keys():
+ ModGuid = ModKey[0]
+ ModVersion = ModKey[1]
+ ModName = ModKey[2]
+ ModInstallPath = ModKey[3]
+ self._AddStandaloneModule(ModGuid, ModVersion, ModName, \
+ DpObj.Header.GetGuid(), \
+ DpObj.Header.GetVersion(), \
+ ModInstallPath)
+ ModObj = DpObj.ModuleSurfaceArea[ModKey]
+ for Dep in ModObj.GetPackageDependencyList():
+ DepexGuid = Dep.GetGuid()
+ DepexVersion = Dep.GetVersion()
+ self._AddModuleDepex(ModGuid, ModVersion, ModName, ModInstallPath, \
+ DepexGuid, DepexVersion)
+ for (Path, Md5Sum) in ModObj.FileList:
+ self._AddDpFilePathList(DpObj.Header.GetGuid(), \
+ DpObj.Header.GetVersion(), \
+ Path, Md5Sum)
+
+ #
+ # add tool/misc files
+ #
+ for (Path, Md5Sum) in DpObj.FileList:
+ self._AddDpFilePathList(DpObj.Header.GetGuid(), \
+ DpObj.Header.GetVersion(), Path, Md5Sum)
+
+ self._AddDp(DpObj.Header.GetGuid(), DpObj.Header.GetVersion(), \
+ NewDpPkgFileName, DpPkgFileName, RePackage)
+
+ except sqlite3.IntegrityError as DetailMsg:
+ Logger.Error("UPT",
+ UPT_DB_UPDATE_ERROR,
+ ST.ERR_UPT_DB_UPDATE_ERROR,
+ ExtraData = DetailMsg
+ )
+
+ ## Add a distribution install information
+ #
+ # @param Guid Guid of the distribution package
+ # @param Version Version of the distribution package
+ # @param NewDpFileName the saved filename of distribution package file
+ # @param DistributionFileName the filename of distribution package file
+ #
+ def _AddDp(self, Guid, Version, NewDpFileName, DistributionFileName, \
+ RePackage):
+
+ if Version is None or len(Version.strip()) == 0:
+ Version = 'N/A'
+
+ #
+ # Add newly installed DP information to DB.
+ #
+ if NewDpFileName is None or len(NewDpFileName.strip()) == 0:
+ PkgFileName = 'N/A'
+ else:
+ PkgFileName = NewDpFileName
+ CurrentTime = time.time()
+ SqlCommand = \
+ """insert into %s values('%s', '%s', %s, '%s', '%s', '%s')""" % \
+ (self.DpTable, Guid, Version, CurrentTime, PkgFileName, \
+ DistributionFileName, str(RePackage).upper())
+ self.Cur.execute(SqlCommand)
+
+
+ ## Add a file list from DP
+ #
+ # @param DpGuid: A DpGuid
+ # @param DpVersion: A DpVersion
+ # @param Path: A Path
+ # @param Path: A Md5Sum
+ #
+ def _AddDpFilePathList(self, DpGuid, DpVersion, Path, Md5Sum):
+ Path = os.path.normpath(Path)
+ if pf.system() == 'Windows':
+ if Path.startswith(self.Workspace):
+ Path = Path[len(self.Workspace):]
+ else:
+ if Path.startswith(self.Workspace + os.sep):
+ Path = Path[len(self.Workspace)+1:]
+ SqlCommand = """insert into %s values('%s', '%s', '%s', '%s')""" % \
+ (self.DpFileListTable, Path, DpGuid, DpVersion, Md5Sum)
+
+ self.Cur.execute(SqlCommand)
+
+ ## Add a package install information
+ #
+ # @param Guid: A package guid
+ # @param Version: A package version
+ # @param DpGuid: A DpGuid
+ # @param DpVersion: A DpVersion
+ # @param Path: A Path
+ #
+ def _AddPackage(self, Guid, Version, DpGuid=None, DpVersion=None, Path=''):
+
+ if Version is None or len(Version.strip()) == 0:
+ Version = 'N/A'
+
+ if DpGuid is None or len(DpGuid.strip()) == 0:
+ DpGuid = 'N/A'
+
+ if DpVersion is None or len(DpVersion.strip()) == 0:
+ DpVersion = 'N/A'
+
+ #
+ # Add newly installed package information to DB.
+ #
+ CurrentTime = time.time()
+ SqlCommand = \
+ """insert into %s values('%s', '%s', %s, '%s', '%s', '%s')""" % \
+ (self.PkgTable, Guid, Version, CurrentTime, DpGuid, DpVersion, Path)
+ self.Cur.execute(SqlCommand)
+
+ ## Add a module that from a package install information
+ #
+ # @param Guid: Module Guid
+ # @param Version: Module version
+ # @param Name: Module Name
+ # @param PkgGuid: Package Guid
+ # @param PkgVersion: Package version
+ # @param Path: Package relative path that module installs
+ #
+ def _AddModuleInPackage(self, Guid, Version, Name, PkgGuid=None, \
+ PkgVersion=None, Path=''):
+
+ if Version is None or len(Version.strip()) == 0:
+ Version = 'N/A'
+
+ if PkgGuid is None or len(PkgGuid.strip()) == 0:
+ PkgGuid = 'N/A'
+
+ if PkgVersion is None or len(PkgVersion.strip()) == 0:
+ PkgVersion = 'N/A'
+
+ if os.name == 'posix':
+ Path = Path.replace('\\', os.sep)
+ else:
+ Path = Path.replace('/', os.sep)
+
+ #
+ # Add module from package information to DB.
+ #
+ CurrentTime = time.time()
+ SqlCommand = \
+ """insert into %s values('%s', '%s', '%s', %s, '%s', '%s', '%s')""" % \
+ (self.ModInPkgTable, Guid, Version, Name, CurrentTime, PkgGuid, PkgVersion, \
+ Path)
+ self.Cur.execute(SqlCommand)
+
+ ## Add a module that is standalone install information
+ #
+ # @param Guid: a module Guid
+ # @param Version: a module Version
+ # @param Name: a module name
+ # @param DpGuid: a DpGuid
+ # @param DpVersion: a DpVersion
+ # @param Path: path
+ #
+ def _AddStandaloneModule(self, Guid, Version, Name, DpGuid=None, \
+ DpVersion=None, Path=''):
+
+ if Version is None or len(Version.strip()) == 0:
+ Version = 'N/A'
+
+ if DpGuid is None or len(DpGuid.strip()) == 0:
+ DpGuid = 'N/A'
+
+ if DpVersion is None or len(DpVersion.strip()) == 0:
+ DpVersion = 'N/A'
+
+ #
+ # Add module standalone information to DB.
+ #
+ CurrentTime = time.time()
+ SqlCommand = \
+ """insert into %s values('%s', '%s', '%s', %s, '%s', '%s', '%s')""" % \
+ (self.StandaloneModTable, Guid, Version, Name, CurrentTime, DpGuid, \
+ DpVersion, Path)
+ self.Cur.execute(SqlCommand)
+
+ ## Add a module depex
+ #
+ # @param Guid: a module Guid
+ # @param Version: a module Version
+ # @param Name: a module name
+ # @param DepexGuid: a module DepexGuid
+ # @param DepexVersion: a module DepexVersion
+ #
+ def _AddModuleDepex(self, Guid, Version, Name, Path, DepexGuid=None, \
+ DepexVersion=None):
+
+ if DepexGuid is None or len(DepexGuid.strip()) == 0:
+ DepexGuid = 'N/A'
+
+ if DepexVersion is None or len(DepexVersion.strip()) == 0:
+ DepexVersion = 'N/A'
+
+ if os.name == 'posix':
+ Path = Path.replace('\\', os.sep)
+ else:
+ Path = Path.replace('/', os.sep)
+
+ #
+ # Add module depex information to DB.
+ #
+ SqlCommand = """insert into %s values('%s', '%s', '%s', '%s', '%s', '%s')"""\
+ % (self.ModDepexTable, Guid, Version, Name, Path, DepexGuid, DepexVersion)
+ self.Cur.execute(SqlCommand)
+
+ ## Remove a distribution install information, if no version specified,
+ # remove all DPs with this Guid.
+ #
+ # @param DpGuid: guid of dpex
+ # @param DpVersion: version of dpex
+ #
+ def RemoveDpObj(self, DpGuid, DpVersion):
+
+ PkgList = self.GetPackageListFromDp(DpGuid, DpVersion)
+ #
+ # delete from ModDepex the standalone module's dependency
+ #
+ SqlCommand = \
+ """delete from ModDepexInfo where ModDepexInfo.ModuleGuid in
+ (select ModuleGuid from StandaloneModInfo as B where B.DpGuid = '%s'
+ and B.DpVersion = '%s')
+ and ModDepexInfo.ModuleVersion in
+ (select ModuleVersion from StandaloneModInfo as B
+ where B.DpGuid = '%s' and B.DpVersion = '%s')
+ and ModDepexInfo.ModuleName in
+ (select ModuleName from StandaloneModInfo as B
+ where B.DpGuid = '%s' and B.DpVersion = '%s')
+ and ModDepexInfo.InstallPath in
+ (select InstallPath from StandaloneModInfo as B
+ where B.DpGuid = '%s' and B.DpVersion = '%s') """ % \
+ (DpGuid, DpVersion, DpGuid, DpVersion, DpGuid, DpVersion, DpGuid, DpVersion)
+
+ self.Cur.execute(SqlCommand)
+ #
+ # delete from ModDepex the from pkg module's dependency
+ #
+ for Pkg in PkgList:
+
+ SqlCommand = \
+ """delete from ModDepexInfo where ModDepexInfo.ModuleGuid in
+ (select ModuleGuid from ModInPkgInfo
+ where ModInPkgInfo.PackageGuid ='%s' and
+ ModInPkgInfo.PackageVersion = '%s')
+ and ModDepexInfo.ModuleVersion in
+ (select ModuleVersion from ModInPkgInfo
+ where ModInPkgInfo.PackageGuid ='%s' and
+ ModInPkgInfo.PackageVersion = '%s')
+ and ModDepexInfo.ModuleName in
+ (select ModuleName from ModInPkgInfo
+ where ModInPkgInfo.PackageGuid ='%s' and
+ ModInPkgInfo.PackageVersion = '%s')
+ and ModDepexInfo.InstallPath in
+ (select InstallPath from ModInPkgInfo where
+ ModInPkgInfo.PackageGuid ='%s'
+ and ModInPkgInfo.PackageVersion = '%s')""" \
+ % (Pkg[0], Pkg[1], Pkg[0], Pkg[1], Pkg[0], Pkg[1], Pkg[0], Pkg[1])
+
+ self.Cur.execute(SqlCommand)
+ #
+ # delete the standalone module
+ #
+ SqlCommand = \
+ """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.StandaloneModTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+ #
+ # delete the from pkg module
+ #
+ for Pkg in PkgList:
+ SqlCommand = \
+ """delete from %s where %s.PackageGuid ='%s'
+ and %s.PackageVersion = '%s'""" % \
+ (self.ModInPkgTable, self.ModInPkgTable, Pkg[0], \
+ self.ModInPkgTable, Pkg[1])
+ self.Cur.execute(SqlCommand)
+ #
+ # delete packages
+ #
+ SqlCommand = \
+ """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.PkgTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+ #
+ # delete file list from DP
+ #
+ SqlCommand = \
+ """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.DpFileListTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+ #
+ # delete DP
+ #
+ SqlCommand = \
+ """delete from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.DpTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ #self.Conn.commit()
+
+ ## Get a list of distribution install information.
+ #
+ # @param Guid: distribution package guid
+ # @param Version: distribution package version
+ #
+ def GetDp(self, Guid, Version):
+
+ if Version is None or len(Version.strip()) == 0:
+ Version = 'N/A'
+ Logger.Verbose(ST.MSG_GET_DP_INSTALL_LIST)
+ (DpGuid, DpVersion) = (Guid, Version)
+ SqlCommand = """select * from %s where DpGuid ='%s'""" % \
+ (self.DpTable, DpGuid)
+ self.Cur.execute(SqlCommand)
+
+ else:
+ Logger.Verbose(ST.MSG_GET_DP_INSTALL_INFO_START)
+ (DpGuid, DpVersion) = (Guid, Version)
+ SqlCommand = \
+ """select * from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.DpTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ DpList = []
+ for DpInfo in self.Cur:
+ DpGuid = DpInfo[0]
+ DpVersion = DpInfo[1]
+ InstallTime = DpInfo[2]
+ PkgFileName = DpInfo[3]
+ DpList.append((DpGuid, DpVersion, InstallTime, PkgFileName))
+
+ Logger.Verbose(ST.MSG_GET_DP_INSTALL_INFO_FINISH)
+ return DpList
+
+ ## Get a list of distribution install dirs
+ #
+ # @param Guid: distribution package guid
+ # @param Version: distribution package version
+ #
+ def GetDpInstallDirList(self, Guid, Version):
+ SqlCommand = """select InstallPath from PkgInfo where DpGuid = '%s' and DpVersion = '%s'""" % (Guid, Version)
+ self.Cur.execute(SqlCommand)
+ DirList = []
+ for Result in self.Cur:
+ if Result[0] not in DirList:
+ DirList.append(Result[0])
+
+ SqlCommand = """select InstallPath from StandaloneModInfo where DpGuid = '%s' and DpVersion = '%s'""" % \
+ (Guid, Version)
+ self.Cur.execute(SqlCommand)
+ for Result in self.Cur:
+ if Result[0] not in DirList:
+ DirList.append(Result[0])
+
+ return DirList
+
+
+ ## Get a list of distribution install file path information.
+ #
+ # @param Guid: distribution package guid
+ # @param Version: distribution package version
+ #
+ def GetDpFileList(self, Guid, Version):
+
+ (DpGuid, DpVersion) = (Guid, Version)
+ SqlCommand = \
+ """select * from %s where DpGuid ='%s' and DpVersion = '%s'""" % \
+ (self.DpFileListTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ PathList = []
+ for Result in self.Cur:
+ Path = Result[0]
+ Md5Sum = Result[3]
+ PathList.append((os.path.join(self.Workspace, Path), Md5Sum))
+
+ return PathList
+
+ ## Get files' repackage attribute if present that are installed into current workspace
+ #
+ # @retval FileDict: a Dict of file, key is file path, value is (DpGuid, DpVersion, NewDpFileName, RePackage)
+ #
+ def GetRePkgDict(self):
+ SqlCommand = """select * from %s """ % (self.DpTable)
+ self.Cur.execute(SqlCommand)
+
+ DpInfoList = []
+ for Result in self.Cur:
+ DpInfoList.append(Result)
+
+ FileDict = {}
+ for Result in DpInfoList:
+ DpGuid = Result[0]
+ DpVersion = Result[1]
+ NewDpFileName = Result[3]
+ RePackage = Result[5]
+ if RePackage == 'TRUE':
+ RePackage = True
+ else:
+ RePackage = False
+ for FileInfo in self.GetDpFileList(DpGuid, DpVersion):
+ PathInfo = FileInfo[0]
+ FileDict[PathInfo] = DpGuid, DpVersion, NewDpFileName, RePackage
+
+ return FileDict
+
+ ## Get (Guid, Version) from distribution file name information.
+ #
+ # @param DistributionFile: Distribution File
+ #
+ def GetDpByName(self, DistributionFile):
+ SqlCommand = """select * from %s where NewPkgFileName = '%s'""" % \
+ (self.DpTable, DistributionFile)
+ self.Cur.execute(SqlCommand)
+
+ for Result in self.Cur:
+ DpGuid = Result[0]
+ DpVersion = Result[1]
+ NewDpFileName = Result[3]
+
+ return (DpGuid, DpVersion, NewDpFileName)
+ else:
+ return (None, None, None)
+
+ ## Get a list of package information.
+ #
+ # @param Guid: package guid
+ # @param Version: package version
+ #
+ def GetPackage(self, Guid, Version, DpGuid='', DpVersion=''):
+
+ if DpVersion == '' or DpGuid == '':
+
+ (PackageGuid, PackageVersion) = (Guid, Version)
+ SqlCommand = """select * from %s where PackageGuid ='%s'
+ and PackageVersion = '%s'""" % (self.PkgTable, PackageGuid, \
+ PackageVersion)
+ self.Cur.execute(SqlCommand)
+
+ elif Version is None or len(Version.strip()) == 0:
+
+ SqlCommand = """select * from %s where PackageGuid ='%s'""" % \
+ (self.PkgTable, Guid)
+ self.Cur.execute(SqlCommand)
+ else:
+ (PackageGuid, PackageVersion) = (Guid, Version)
+ SqlCommand = """select * from %s where PackageGuid ='%s' and
+ PackageVersion = '%s'
+ and DpGuid = '%s' and DpVersion = '%s'""" % \
+ (self.PkgTable, PackageGuid, PackageVersion, \
+ DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ PkgList = []
+ for PkgInfo in self.Cur:
+ PkgGuid = PkgInfo[0]
+ PkgVersion = PkgInfo[1]
+ InstallTime = PkgInfo[2]
+ InstallPath = PkgInfo[5]
+ PkgList.append((PkgGuid, PkgVersion, InstallTime, DpGuid, \
+ DpVersion, InstallPath))
+
+ return PkgList
+
+
+ ## Get a list of module in package information.
+ #
+ # @param Guid: A module guid
+ # @param Version: A module version
+ #
+ def GetModInPackage(self, Guid, Version, Name, Path, PkgGuid='', PkgVersion=''):
+ (ModuleGuid, ModuleVersion, ModuleName, InstallPath) = (Guid, Version, Name, Path)
+ if PkgVersion == '' or PkgGuid == '':
+ SqlCommand = """select * from %s where ModuleGuid ='%s' and
+ ModuleVersion = '%s' and InstallPath = '%s'
+ and ModuleName = '%s'""" % (self.ModInPkgTable, ModuleGuid, \
+ ModuleVersion, InstallPath, ModuleName)
+ self.Cur.execute(SqlCommand)
+ else:
+ SqlCommand = """select * from %s where ModuleGuid ='%s' and
+ ModuleVersion = '%s' and InstallPath = '%s'
+ and ModuleName = '%s' and PackageGuid ='%s'
+ and PackageVersion = '%s'
+ """ % (self.ModInPkgTable, ModuleGuid, \
+ ModuleVersion, InstallPath, ModuleName, PkgGuid, PkgVersion)
+ self.Cur.execute(SqlCommand)
+
+ ModList = []
+ for ModInfo in self.Cur:
+ ModGuid = ModInfo[0]
+ ModVersion = ModInfo[1]
+ InstallTime = ModInfo[2]
+ InstallPath = ModInfo[5]
+ ModList.append((ModGuid, ModVersion, InstallTime, PkgGuid, \
+ PkgVersion, InstallPath))
+
+ return ModList
+
+ ## Get a list of module standalone.
+ #
+ # @param Guid: A module guid
+ # @param Version: A module version
+ #
+ def GetStandaloneModule(self, Guid, Version, Name, Path, DpGuid='', DpVersion=''):
+ (ModuleGuid, ModuleVersion, ModuleName, InstallPath) = (Guid, Version, Name, Path)
+ if DpGuid == '':
+ SqlCommand = """select * from %s where ModuleGuid ='%s' and
+ ModuleVersion = '%s' and InstallPath = '%s'
+ and ModuleName = '%s'""" % (self.StandaloneModTable, ModuleGuid, \
+ ModuleVersion, InstallPath, ModuleName)
+ self.Cur.execute(SqlCommand)
+
+ else:
+ SqlCommand = """select * from %s where ModuleGuid ='%s' and
+ ModuleVersion = '%s' and InstallPath = '%s' and ModuleName = '%s' and DpGuid ='%s' and DpVersion = '%s'
+ """ % (self.StandaloneModTable, ModuleGuid, \
+ ModuleVersion, ModuleName, InstallPath, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ ModList = []
+ for ModInfo in self.Cur:
+ ModGuid = ModInfo[0]
+ ModVersion = ModInfo[1]
+ InstallTime = ModInfo[2]
+ InstallPath = ModInfo[5]
+ ModList.append((ModGuid, ModVersion, InstallTime, DpGuid, \
+ DpVersion, InstallPath))
+
+ return ModList
+
+ ## Get a list of module information that comes from DP.
+ #
+ # @param DpGuid: A Distribution Guid
+ # @param DpVersion: A Distribution version
+ #
+ def GetSModInsPathListFromDp(self, DpGuid, DpVersion):
+
+ PathList = []
+ SqlCommand = """select InstallPath from %s where DpGuid ='%s'
+ and DpVersion = '%s'
+ """ % (self.StandaloneModTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ for Result in self.Cur:
+ InstallPath = Result[0]
+ PathList.append(InstallPath)
+
+ return PathList
+
+ ## Get a list of package information.
+ #
+ # @param DpGuid: A Distribution Guid
+ # @param DpVersion: A Distribution version
+ #
+ def GetPackageListFromDp(self, DpGuid, DpVersion):
+
+ SqlCommand = """select * from %s where DpGuid ='%s' and
+ DpVersion = '%s' """ % (self.PkgTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+
+ PkgList = []
+ for PkgInfo in self.Cur:
+ PkgGuid = PkgInfo[0]
+ PkgVersion = PkgInfo[1]
+ InstallPath = PkgInfo[5]
+ PkgList.append((PkgGuid, PkgVersion, InstallPath))
+
+ return PkgList
+
+ ## Get a list of modules that depends on package information from a DP.
+ #
+ # @param DpGuid: A Distribution Guid
+ # @param DpVersion: A Distribution version
+ #
+ def GetDpDependentModuleList(self, DpGuid, DpVersion):
+
+ ModList = []
+ PkgList = self.GetPackageListFromDp(DpGuid, DpVersion)
+ if len(PkgList) > 0:
+ return ModList
+
+ for Pkg in PkgList:
+ #
+ # get all in-package modules that depends on current
+ # Pkg (Guid match, Version match or NA) but not belong to
+ # current Pkg
+ #
+ SqlCommand = """select t1.ModuleGuid, t1.ModuleVersion,
+ t1.InstallPath from %s as t1, %s as t2 where
+ t1.ModuleGuid = t2.ModuleGuid and
+ t1.ModuleVersion = t2.ModuleVersion and t2.DepexGuid ='%s'
+ and (t2.DepexVersion = '%s' or t2.DepexVersion = 'N/A') and
+ t1.PackageGuid != '%s' and t1.PackageVersion != '%s'
+ """ % (self.ModInPkgTable, \
+ self.ModDepexTable, Pkg[0], Pkg[1], Pkg[0], \
+ Pkg[1])
+ self.Cur.execute(SqlCommand)
+ for ModInfo in self.Cur:
+ ModGuid = ModInfo[0]
+ ModVersion = ModInfo[1]
+ InstallPath = ModInfo[2]
+ ModList.append((ModGuid, ModVersion, InstallPath))
+
+ #
+ # get all modules from standalone modules that depends on current
+ #Pkg (Guid match, Version match or NA) but not in current dp
+ #
+ SqlCommand = \
+ """select t1.ModuleGuid, t1.ModuleVersion, t1.InstallPath
+ from %s as t1, %s as t2 where t1.ModuleGuid = t2.ModuleGuid and
+ t1.ModuleVersion = t2.ModuleVersion and t2.DepexGuid ='%s'
+ and (t2.DepexVersion = '%s' or t2.DepexVersion = 'N/A') and
+ t1.DpGuid != '%s' and t1.DpVersion != '%s'
+ """ % \
+ (self.StandaloneModTable, self.ModDepexTable, Pkg[0], \
+ Pkg[1], DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+ for ModInfo in self.Cur:
+ ModGuid = ModInfo[0]
+ ModVersion = ModInfo[1]
+ InstallPath = ModInfo[2]
+ ModList.append((ModGuid, ModVersion, InstallPath))
+
+
+ return ModList
+
+ ## Get Dp's list of modules.
+ #
+ # @param DpGuid: A Distribution Guid
+ # @param DpVersion: A Distribution version
+ #
+ def GetDpModuleList(self, DpGuid, DpVersion):
+ ModList = []
+ #
+ # get Dp module list from the DpFileList table
+ #
+ SqlCommand = """select FilePath
+ from %s
+ where DpGuid = '%s' and DpVersion = '%s' and
+ FilePath like '%%.inf'
+ """ % (self.DpFileListTable, DpGuid, DpVersion)
+ self.Cur.execute(SqlCommand)
+ for ModuleInfo in self.Cur:
+ FilePath = ModuleInfo[0]
+ ModList.append(os.path.join(self.Workspace, FilePath))
+
+ return ModList
+
+
+ ## Get a module depex
+ #
+ # @param DpGuid: A module Guid
+ # @param DpVersion: A module version
+ # @param Path:
+ #
+ def GetModuleDepex(self, Guid, Version, Path):
+
+ #
+ # Get module depex information to DB.
+ #
+ SqlCommand = """select * from %s where ModuleGuid ='%s' and
+ ModuleVersion = '%s' and InstallPath ='%s'
+ """ % (self.ModDepexTable, Guid, Version, Path)
+ self.Cur.execute(SqlCommand)
+
+
+ DepexList = []
+ for DepInfo in self.Cur:
+ DepexGuid = DepInfo[3]
+ DepexVersion = DepInfo[4]
+ DepexList.append((DepexGuid, DepexVersion))
+
+ return DepexList
+
+ ## Inventory the distribution installed to current workspace
+ #
+ # Inventory the distribution installed to current workspace
+ #
+ def InventoryDistInstalled(self):
+ SqlCommand = """select * from %s """ % (self.DpTable)
+ self.Cur.execute(SqlCommand)
+
+ DpInfoList = []
+ for Result in self.Cur:
+ DpGuid = Result[0]
+ DpVersion = Result[1]
+ DpAliasName = Result[3]
+ DpFileName = Result[4]
+ DpInfoList.append((DpGuid, DpVersion, DpFileName, DpAliasName))
+
+ return DpInfoList
+
+ ## Close entire database
+ #
+ # Close the connection and cursor
+ #
+ def CloseDb(self):
+ #
+ # drop the dummy table
+ #
+ SqlCommand = """
+ drop table IF EXISTS %s
+ """ % self.DummyTable
+ self.Cur.execute(SqlCommand)
+ self.Conn.commit()
+
+ self.Cur.close()
+ self.Conn.close()
+
+ ## Convert To Sql String
+ #
+ # 1. Replace "'" with "''" in each item of StringList
+ #
+ # @param StringList: A list for strings to be converted
+ #
+ def __ConvertToSqlString(self, StringList):
+ if self.DpTable:
+ pass
+ return list(map(lambda s: s.replace("'", "''"), StringList))
+
+
+
+
diff --git a/BaseTools/Source/Python/UPT/Core/PackageFile.py b/BaseTools/Source/Python/UPT/Core/PackageFile.py
index c9157e84d6..c45ca7d78e 100644
--- a/BaseTools/Source/Python/UPT/Core/PackageFile.py
+++ b/BaseTools/Source/Python/UPT/Core/PackageFile.py
@@ -1,250 +1,250 @@
-## @file
-#
-# PackageFile class represents the zip file of a distribution package.
-#
-# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-#
-
-'''
-PackageFile
-'''
-
-##
-# Import Modules
-#
-import os.path
-import zipfile
-import tempfile
-import platform
-
-from Logger.ToolError import FILE_OPEN_FAILURE
-from Logger.ToolError import FILE_CHECKSUM_FAILURE
-from Logger.ToolError import FILE_NOT_FOUND
-from Logger.ToolError import FILE_DECOMPRESS_FAILURE
-from Logger.ToolError import FILE_UNKNOWN_ERROR
-from Logger.ToolError import FILE_WRITE_FAILURE
-from Logger.ToolError import FILE_COMPRESS_FAILURE
-import Logger.Log as Logger
-from Logger import StringTable as ST
-from Library.Misc import CreateDirectory
-from Library.Misc import RemoveDirectory
-from Core.FileHook import __FileHookOpen__
-from Common.MultipleWorkspace import MultipleWorkspace as mws
-
-
-class PackageFile:
- def __init__(self, FileName, Mode="r"):
- self._FileName = FileName
- if Mode not in ["r", "w", "a"]:
- Mode = "r"
- try:
- self._ZipFile = zipfile.ZipFile(FileName, Mode, \
- zipfile.ZIP_DEFLATED)
- self._Files = {}
- for Filename in self._ZipFile.namelist():
- self._Files[os.path.normpath(Filename)] = Filename
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_OPEN_FAILURE,
- ExtraData="%s (%s)" % (FileName, str(Xstr)))
-
- BadFile = self._ZipFile.testzip()
- if BadFile is not None:
- Logger.Error("PackagingTool", FILE_CHECKSUM_FAILURE,
- ExtraData="[%s] in %s" % (BadFile, FileName))
-
- def GetZipFile(self):
- return self._ZipFile
-
- ## Get file name
- #
- def __str__(self):
- return self._FileName
-
- ## Extract the file
- #
- # @param To: the destination file
- #
- def Unpack(self, ToDest):
- for FileN in self._ZipFile.namelist():
- ToFile = os.path.normpath(os.path.join(ToDest, FileN))
- Msg = "%s -> %s" % (FileN, ToFile)
- Logger.Info(Msg)
- self.Extract(FileN, ToFile)
-
- ## Extract the file
- #
- # @param File: the extracted file
- # @param ToFile: the destination file
- #
- def UnpackFile(self, File, ToFile):
- File = File.replace('\\', '/')
- if File in self._ZipFile.namelist():
- Msg = "%s -> %s" % (File, ToFile)
- Logger.Info(Msg)
- self.Extract(File, ToFile)
- return ToFile
-
- return ''
-
- ## Extract the file
- #
- # @param Which: the source path
- # @param ToDest: the destination path
- #
- def Extract(self, Which, ToDest):
- Which = os.path.normpath(Which)
- if Which not in self._Files:
- Logger.Error("PackagingTool", FILE_NOT_FOUND,
- ExtraData="[%s] in %s" % (Which, self._FileName))
- try:
- FileContent = self._ZipFile.read(self._Files[Which])
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_DECOMPRESS_FAILURE,
- ExtraData="[%s] in %s (%s)" % (Which, \
- self._FileName, \
- str(Xstr)))
- try:
- CreateDirectory(os.path.dirname(ToDest))
- if os.path.exists(ToDest) and not os.access(ToDest, os.W_OK):
- Logger.Warn("PackagingTool", \
- ST.WRN_FILE_NOT_OVERWRITTEN % ToDest)
- return
- else:
- ToFile = __FileHookOpen__(ToDest, 'wb')
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_OPEN_FAILURE,
- ExtraData="%s (%s)" % (ToDest, str(Xstr)))
-
- try:
- ToFile.write(FileContent)
- ToFile.close()
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_WRITE_FAILURE,
- ExtraData="%s (%s)" % (ToDest, str(Xstr)))
-
- ## Remove the file
- #
- # @param Files: the removed files
- #
- def Remove(self, Files):
- TmpDir = os.path.join(tempfile.gettempdir(), ".packaging")
- if os.path.exists(TmpDir):
- RemoveDirectory(TmpDir, True)
-
- os.mkdir(TmpDir)
- self.Unpack(TmpDir)
- for SinF in Files:
- SinF = os.path.normpath(SinF)
- if SinF not in self._Files:
- Logger.Error("PackagingTool", FILE_NOT_FOUND,
- ExtraData="%s is not in %s!" % \
- (SinF, self._FileName))
- self._Files.pop(SinF)
- self._ZipFile.close()
-
- self._ZipFile = zipfile.ZipFile(self._FileName, "w", \
- zipfile.ZIP_DEFLATED)
- Cwd = os.getcwd()
- os.chdir(TmpDir)
- self.PackFiles(self._Files)
- os.chdir(Cwd)
- RemoveDirectory(TmpDir, True)
-
- ## Pack the files under Top directory, the directory shown in the zipFile start from BaseDir,
- # BaseDir should be the parent directory of the Top directory, for example,
- # Pack(Workspace\Dir1, Workspace) will pack files under Dir1, and the path in the zipfile will
- # start from Workspace
- #
- # @param Top: the top directory
- # @param BaseDir: the base directory
- #
- def Pack(self, Top, BaseDir):
- if not os.path.isdir(Top):
- Logger.Error("PackagingTool", FILE_UNKNOWN_ERROR, \
- "%s is not a directory!" %Top)
-
- FilesToPack = []
- Cwd = os.getcwd()
- os.chdir(BaseDir)
- RelaDir = Top[Top.upper().find(BaseDir.upper()).\
- join(len(BaseDir).join(1)):]
-
- for Root, Dirs, Files in os.walk(RelaDir):
- if 'CVS' in Dirs:
- Dirs.remove('CVS')
- if '.svn' in Dirs:
- Dirs.remove('.svn')
-
- for Dir in Dirs:
- if Dir.startswith('.'):
- Dirs.remove(Dir)
- for File1 in Files:
- if File1.startswith('.'):
- continue
- ExtName = os.path.splitext(File1)[1]
- #
- # skip '.dec', '.inf', '.dsc', '.fdf' files
- #
- if ExtName.lower() in ['.dec', '.inf', '.dsc', '.fdf']:
- continue
- FilesToPack.append(os.path.join(Root, File1))
- self.PackFiles(FilesToPack)
- os.chdir(Cwd)
-
- ## Pack the file
- #
- # @param Files: the files to pack
- #
- def PackFiles(self, Files):
- for File in Files:
- Cwd = os.getcwd()
- os.chdir(mws.getWs(mws.WORKSPACE, File))
- self.PackFile(File)
- os.chdir(Cwd)
-
- ## Pack the file
- #
- # @param File: the files to pack
- # @param ArcName: the Arc Name
- #
- def PackFile(self, File, ArcName=None):
- try:
- #
- # avoid packing same file multiple times
- #
- if platform.system() != 'Windows':
- File = File.replace('\\', '/')
- ZipedFilesNameList = self._ZipFile.namelist()
- for ZipedFile in ZipedFilesNameList:
- if File == os.path.normpath(ZipedFile):
- return
- Logger.Info("packing ..." + File)
- self._ZipFile.write(File, ArcName)
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_COMPRESS_FAILURE,
- ExtraData="%s (%s)" % (File, str(Xstr)))
-
- ## Write data to the packed file
- #
- # @param Data: data to write
- # @param ArcName: the Arc Name
- #
- def PackData(self, Data, ArcName):
- try:
- if os.path.splitext(ArcName)[1].lower() == '.pkg':
- Data = Data.encode('utf_8')
- self._ZipFile.writestr(ArcName, Data)
- except BaseException as Xstr:
- Logger.Error("PackagingTool", FILE_COMPRESS_FAILURE,
- ExtraData="%s (%s)" % (ArcName, str(Xstr)))
-
- ## Close file
- #
- #
- def Close(self):
- self._ZipFile.close()
-
-
-
+## @file
+#
+# PackageFile class represents the zip file of a distribution package.
+#
+# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+#
+
+'''
+PackageFile
+'''
+
+##
+# Import Modules
+#
+import os.path
+import zipfile
+import tempfile
+import platform
+
+from Logger.ToolError import FILE_OPEN_FAILURE
+from Logger.ToolError import FILE_CHECKSUM_FAILURE
+from Logger.ToolError import FILE_NOT_FOUND
+from Logger.ToolError import FILE_DECOMPRESS_FAILURE
+from Logger.ToolError import FILE_UNKNOWN_ERROR
+from Logger.ToolError import FILE_WRITE_FAILURE
+from Logger.ToolError import FILE_COMPRESS_FAILURE
+import Logger.Log as Logger
+from Logger import StringTable as ST
+from Library.Misc import CreateDirectory
+from Library.Misc import RemoveDirectory
+from Core.FileHook import __FileHookOpen__
+from Common.MultipleWorkspace import MultipleWorkspace as mws
+
+
+class PackageFile:
+ def __init__(self, FileName, Mode="r"):
+ self._FileName = FileName
+ if Mode not in ["r", "w", "a"]:
+ Mode = "r"
+ try:
+ self._ZipFile = zipfile.ZipFile(FileName, Mode, \
+ zipfile.ZIP_DEFLATED)
+ self._Files = {}
+ for Filename in self._ZipFile.namelist():
+ self._Files[os.path.normpath(Filename)] = Filename
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_OPEN_FAILURE,
+ ExtraData="%s (%s)" % (FileName, str(Xstr)))
+
+ BadFile = self._ZipFile.testzip()
+ if BadFile is not None:
+ Logger.Error("PackagingTool", FILE_CHECKSUM_FAILURE,
+ ExtraData="[%s] in %s" % (BadFile, FileName))
+
+ def GetZipFile(self):
+ return self._ZipFile
+
+ ## Get file name
+ #
+ def __str__(self):
+ return self._FileName
+
+ ## Extract the file
+ #
+ # @param To: the destination file
+ #
+ def Unpack(self, ToDest):
+ for FileN in self._ZipFile.namelist():
+ ToFile = os.path.normpath(os.path.join(ToDest, FileN))
+ Msg = "%s -> %s" % (FileN, ToFile)
+ Logger.Info(Msg)
+ self.Extract(FileN, ToFile)
+
+ ## Extract the file
+ #
+ # @param File: the extracted file
+ # @param ToFile: the destination file
+ #
+ def UnpackFile(self, File, ToFile):
+ File = File.replace('\\', '/')
+ if File in self._ZipFile.namelist():
+ Msg = "%s -> %s" % (File, ToFile)
+ Logger.Info(Msg)
+ self.Extract(File, ToFile)
+ return ToFile
+
+ return ''
+
+ ## Extract the file
+ #
+ # @param Which: the source path
+ # @param ToDest: the destination path
+ #
+ def Extract(self, Which, ToDest):
+ Which = os.path.normpath(Which)
+ if Which not in self._Files:
+ Logger.Error("PackagingTool", FILE_NOT_FOUND,
+ ExtraData="[%s] in %s" % (Which, self._FileName))
+ try:
+ FileContent = self._ZipFile.read(self._Files[Which])
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_DECOMPRESS_FAILURE,
+ ExtraData="[%s] in %s (%s)" % (Which, \
+ self._FileName, \
+ str(Xstr)))
+ try:
+ CreateDirectory(os.path.dirname(ToDest))
+ if os.path.exists(ToDest) and not os.access(ToDest, os.W_OK):
+ Logger.Warn("PackagingTool", \
+ ST.WRN_FILE_NOT_OVERWRITTEN % ToDest)
+ return
+ else:
+ ToFile = __FileHookOpen__(ToDest, 'wb')
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_OPEN_FAILURE,
+ ExtraData="%s (%s)" % (ToDest, str(Xstr)))
+
+ try:
+ ToFile.write(FileContent)
+ ToFile.close()
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_WRITE_FAILURE,
+ ExtraData="%s (%s)" % (ToDest, str(Xstr)))
+
+ ## Remove the file
+ #
+ # @param Files: the removed files
+ #
+ def Remove(self, Files):
+ TmpDir = os.path.join(tempfile.gettempdir(), ".packaging")
+ if os.path.exists(TmpDir):
+ RemoveDirectory(TmpDir, True)
+
+ os.mkdir(TmpDir)
+ self.Unpack(TmpDir)
+ for SinF in Files:
+ SinF = os.path.normpath(SinF)
+ if SinF not in self._Files:
+ Logger.Error("PackagingTool", FILE_NOT_FOUND,
+ ExtraData="%s is not in %s!" % \
+ (SinF, self._FileName))
+ self._Files.pop(SinF)
+ self._ZipFile.close()
+
+ self._ZipFile = zipfile.ZipFile(self._FileName, "w", \
+ zipfile.ZIP_DEFLATED)
+ Cwd = os.getcwd()
+ os.chdir(TmpDir)
+ self.PackFiles(self._Files)
+ os.chdir(Cwd)
+ RemoveDirectory(TmpDir, True)
+
+ ## Pack the files under Top directory, the directory shown in the zipFile start from BaseDir,
+ # BaseDir should be the parent directory of the Top directory, for example,
+ # Pack(Workspace\Dir1, Workspace) will pack files under Dir1, and the path in the zipfile will
+ # start from Workspace
+ #
+ # @param Top: the top directory
+ # @param BaseDir: the base directory
+ #
+ def Pack(self, Top, BaseDir):
+ if not os.path.isdir(Top):
+ Logger.Error("PackagingTool", FILE_UNKNOWN_ERROR, \
+ "%s is not a directory!" %Top)
+
+ FilesToPack = []
+ Cwd = os.getcwd()
+ os.chdir(BaseDir)
+ RelaDir = Top[Top.upper().find(BaseDir.upper()).\
+ join(len(BaseDir).join(1)):]
+
+ for Root, Dirs, Files in os.walk(RelaDir):
+ if 'CVS' in Dirs:
+ Dirs.remove('CVS')
+ if '.svn' in Dirs:
+ Dirs.remove('.svn')
+
+ for Dir in Dirs:
+ if Dir.startswith('.'):
+ Dirs.remove(Dir)
+ for File1 in Files:
+ if File1.startswith('.'):
+ continue
+ ExtName = os.path.splitext(File1)[1]
+ #
+ # skip '.dec', '.inf', '.dsc', '.fdf' files
+ #
+ if ExtName.lower() in ['.dec', '.inf', '.dsc', '.fdf']:
+ continue
+ FilesToPack.append(os.path.join(Root, File1))
+ self.PackFiles(FilesToPack)
+ os.chdir(Cwd)
+
+ ## Pack the file
+ #
+ # @param Files: the files to pack
+ #
+ def PackFiles(self, Files):
+ for File in Files:
+ Cwd = os.getcwd()
+ os.chdir(mws.getWs(mws.WORKSPACE, File))
+ self.PackFile(File)
+ os.chdir(Cwd)
+
+ ## Pack the file
+ #
+ # @param File: the files to pack
+ # @param ArcName: the Arc Name
+ #
+ def PackFile(self, File, ArcName=None):
+ try:
+ #
+ # avoid packing same file multiple times
+ #
+ if platform.system() != 'Windows':
+ File = File.replace('\\', '/')
+ ZipedFilesNameList = self._ZipFile.namelist()
+ for ZipedFile in ZipedFilesNameList:
+ if File == os.path.normpath(ZipedFile):
+ return
+ Logger.Info("packing ..." + File)
+ self._ZipFile.write(File, ArcName)
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_COMPRESS_FAILURE,
+ ExtraData="%s (%s)" % (File, str(Xstr)))
+
+ ## Write data to the packed file
+ #
+ # @param Data: data to write
+ # @param ArcName: the Arc Name
+ #
+ def PackData(self, Data, ArcName):
+ try:
+ if os.path.splitext(ArcName)[1].lower() == '.pkg':
+ Data = Data.encode('utf_8')
+ self._ZipFile.writestr(ArcName, Data)
+ except BaseException as Xstr:
+ Logger.Error("PackagingTool", FILE_COMPRESS_FAILURE,
+ ExtraData="%s (%s)" % (ArcName, str(Xstr)))
+
+ ## Close file
+ #
+ #
+ def Close(self):
+ self._ZipFile.close()
+
+
+
diff --git a/BaseTools/Source/Python/UPT/Core/__init__.py b/BaseTools/Source/Python/UPT/Core/__init__.py
index 9af96f9c3c..1b9f9b3226 100644
--- a/BaseTools/Source/Python/UPT/Core/__init__.py
+++ b/BaseTools/Source/Python/UPT/Core/__init__.py
@@ -1,14 +1,14 @@
-## @file
-# Python 'Library' package initialization file.
-#
-# This file is required to make Python interpreter treat the directory
-# as containing package.
-#
-# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
-#
-# SPDX-License-Identifier: BSD-2-Clause-Patent
-#
-
-'''
-Core init file
-'''
+## @file
+# Python 'Library' package initialization file.
+#
+# This file is required to make Python interpreter treat the directory
+# as containing package.
+#
+# Copyright (c) 2011 - 2018, Intel Corporation. All rights reserved.<BR>
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+#
+
+'''
+Core init file
+'''