/
zcml.py
129 lines (106 loc) · 4.85 KB
/
zcml.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
from ftw.upgrade.directory.scanner import Scanner
from ftw.upgrade.directory.wrapper import wrap_upgrade_step
from ftw.upgrade.exceptions import UpgradeStepConfigurationError
from operator import attrgetter
from Products.CMFPlone.interfaces import IMigratingPloneSiteRoot
from Products.GenericSetup.interfaces import EXTENSION
from Products.GenericSetup.interfaces import IProfile
from Products.GenericSetup.registry import _profile_registry
from Products.GenericSetup.registry import GlobalRegistryStorage
from Products.GenericSetup.upgrade import _registerUpgradeStep
from Products.GenericSetup.upgrade import _upgrade_registry
from Products.GenericSetup.upgrade import UpgradeStep
from zope.configuration.fields import Path
from zope.configuration.fields import Tokens
from zope.interface import Interface
import os
import zope.schema
class IUpgradeStepDirectoryDirective(Interface):
profile = zope.schema.TextLine(
title=u"GenericSetup profile id",
required=True)
directory = Path(
title=u'Path to the upgrade steps directory',
required=True)
soft_dependencies = Tokens(
title=u'List of Generic Setup profile dependencies.',
description=u'Format: "my.package:default"',
required=False,
value_type=zope.schema.TextLine())
def upgrade_step_directory_handler(context, profile, directory,
soft_dependencies=None):
dottedname = context.package.__name__
package_dir = os.path.dirname(context.package.__file__)
if package_dir != os.path.abspath(directory):
dottedname += '.' + '.'.join(
os.path.relpath(os.path.abspath(directory), package_dir)
.split(os.sep))
context.action(
discriminator=('upgrade-step:directory', profile),
callable=upgrade_step_directory_action,
args=(profile, dottedname, context.path(directory),
soft_dependencies))
def upgrade_step_directory_action(profile, dottedname, path,
soft_dependencies):
start_version = find_start_version(profile)
scanner = Scanner(dottedname, path)
if profile not in _profile_registry.listProfiles():
raise UpgradeStepConfigurationError(
'The profile "{0}" needs to be registered before registering its'
' upgrade step directory.'.format(profile))
profileinfo = _profile_registry.getProfileInfo(profile)
if profileinfo.get('version', None) is not None:
raise UpgradeStepConfigurationError(
'Registering an upgrades directory for "{0}" requires this profile'
' to not define a version in its metadata.xml.'
' The version is automatically set to the latest upgrade.'.format(
profile))
_package, profilename = profile.split(':', 1)
last_version = ''.join(find_start_version(profile))
for upgrade_info in scanner.scan():
upgrade_profile_name = '{0}-upgrade-{1}'.format(
profilename, upgrade_info['target-version'])
upgrade_handler = wrap_upgrade_step(
handler=upgrade_info['callable'],
upgrade_profile='profile-{0}:{1}'.format(dottedname,
upgrade_profile_name),
base_profile=profile,
target_version=upgrade_info['target-version'])
step = UpgradeStep(upgrade_info['title'],
profile,
upgrade_info['source-version'] or start_version,
upgrade_info['target-version'],
'',
upgrade_handler)
_registerUpgradeStep(step)
_profile_registry.registerProfile(
name=upgrade_profile_name,
title='Upgrade {0} to {1}: {2}'.format(
profile,
upgrade_info['target-version'],
upgrade_info['title']),
description='',
path=upgrade_info['path'],
product=dottedname,
profile_type=EXTENSION,
for_=IMigratingPloneSiteRoot)
last_version = upgrade_info['target-version']
profile = GlobalRegistryStorage(IProfile).get(profile)
profile['version'] = last_version
profile['ftw.upgrade:dependencies'] = soft_dependencies
def find_start_version(profile):
upgrades = _upgrade_registry.getUpgradeStepsForProfile(profile).values()
dests = set()
for upgrade in upgrades:
if isinstance(upgrade, list):
# Those are combined upgrade steps
# (registered with the upgradeSteps directive)
[dests.update(partial_upgrade[1].dest) for partial_upgrade in upgrade]
else:
# Those are simple upgrade steps
# (registered with the upgradeStep directive)
dests.update(upgrade.dest)
if dests:
return max(dests)
else:
return str(10 ** 13)