random_id = random.randrange(1, 9999)

    return random_id, json.dumps({

        "id": random_id,

        "api_key": apikey,

        "method": method,

        "args": kw

    })

jsonify = lambda obj: json.loads(json.dumps(obj))

def crash(*args, **kwargs):

    raise Exception('Total Crash !')

def api_call(test_obj, params):

    response = test_obj.app.post(API_URL, content_type='application/json',

                                 params=params)

    return response

## helpers

def make_user_group(name=TEST_USER_GROUP):

    gr = fixture.create_user_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    UserGroupModel().add_user_to_group(user_group=gr,

                                       user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

def make_repo_group(name=TEST_REPO_GROUP):

    gr = fixture.create_repo_group(name, cur_user=TEST_USER_ADMIN_LOGIN)

    Session().commit()

    return gr

class BaseTestApi(object):

    REPO = None

    REPO_TYPE = None

    @classmethod

    def setup_class(cls):

        cls.usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        cls.apikey = cls.usr.api_key

        cls.test_user = UserModel().create_or_update(

            username='test-api',

            password='test',

            firstname='first',

            lastname='last'

        )

        Session().commit()

        cls.TEST_USER_LOGIN = cls.test_user.username

        cls.apikey_regular = cls.test_user.api_key

    @classmethod

    def teardown_class(cls):

        pass

    def setUp(self):

        self.maxDiff = None

        make_user_group()

        make_repo_group()

    def tearDown(self):

        fixture.destroy_user_group(TEST_USER_GROUP)

        fixture.destroy_gists()

        fixture.destroy_repo_group(TEST_REPO_GROUP)

    def _compare_ok(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': None,

            'result': expected

        })

        given = json.loads(given)

        self.assertEqual(expected, given)

    def _compare_error(self, id_, expected, given):

        expected = jsonify({

            'id': id_,

            'error': expected,

            'result': None

        })

        given = json.loads(given)

        self.assertEqual(expected, given)

    def test_Optional_object(self):

        from rhodecode.controllers.api.api import Optional

        option1 = Optional(None)

        self.assertEqual('<Optional:%s>' % None, repr(option1))

        self.assertEqual(option1(), None)

        self.assertEqual(1, Optional.extract(Optional(1)))

        self.assertEqual('trololo', Optional.extract('trololo'))

        usr = UserModel().get_by_username(username)

        ret = dict(

            msg='created new user `%s`' % username,

            user=jsonify(usr.get_api_data())

        )

        try:

            expected = ret

            self._compare_ok(id_, expected, given=response.body)

        finally:

            fixture.destroy_user(usr.user_id)

    def test_api_create_user_with_extern_name(self):

        username = 'test_new_api_user_passwordless'

        email = username + "@foo.com"

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=username,

                                  email=email, extern_name='rhodecode')

        response = api_call(self, params)

        usr = UserModel().get_by_username(username)

        ret = dict(

            msg='created new user `%s`' % username,

            user=jsonify(usr.get_api_data())

        )

        try:

            expected = ret

            self._compare_ok(id_, expected, given=response.body)

        finally:

            fixture.destroy_user(usr.user_id)

    @mock.patch.object(UserModel, 'create_or_update', crash)

    def test_api_create_user_when_exception_happened(self):

        username = 'test_new_api_user'

        email = username + "@foo.com"

        id_, params = _build_data(self.apikey, 'create_user',

                                  username=username,

                                  email=email,

                                  password='trololo')

        response = api_call(self, params)

        expected = 'failed to create user `%s`' % username

        self._compare_error(id_, expected, given=response.body)

    def test_api_delete_user(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                           firstname=u'u1', lastname=u'u1')

        Session().commit()

        username = usr.username

        email = usr.email

        usr_id = usr.user_id

        ## DELETE THIS USER NOW

        id_, params = _build_data(self.apikey, 'delete_user',

                                  userid=username, )

        response = api_call(self, params)

        ret = {'msg': 'deleted user ID:%s %s' % (usr_id, username),

               'user': None}

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(UserModel, 'delete', crash)

    def test_api_delete_user_when_exception_happened(self):

        usr = UserModel().create_or_update(username=u'test_user',

                                           password=u'qweqwe',

                                           firstname=u'u1', lastname=u'u1')

        Session().commit()

        username = usr.username

        id_, params = _build_data(self.apikey, 'delete_user',

                                  userid=username, )

        response = api_call(self, params)

        ret = 'failed to delete user ID:%s %s' % (usr.user_id,

                                                  usr.username)

        expected = ret

        self._compare_error(id_, expected, given=response.body)

    @parameterized.expand([('firstname', 'new_username'),

                           ('lastname', 'new_username'),

                           ('email', 'new_username'),

                           ('admin', True),

                           ('admin', False),

                           ('extern_type', 'ldap'),

                           ('extern_type', None),

                           ('extern_name', 'test'),

                           ('extern_name', None),

                           ('active', False),

                           ('active', True),

                           ('password', 'newpass')

    ])

    def test_api_update_user(self, name, expected):

        usr = UserModel().get_by_username(self.TEST_USER_LOGIN)

        kw = {name: expected,

              'userid': usr.user_id}

        id_, params = _build_data(self.apikey, 'update_user', **kw)

        response = api_call(self, params)

        ret = {

            'msg': 'updated user ID:%s %s' % (

                usr.user_id, self.TEST_USER_LOGIN),

            'user': jsonify(UserModel() \

                .get_by_username(self.TEST_USER_LOGIN) \

                .get_api_data())

        }

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_update_user_no_changed_params(self):

        usr = UserModel().get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = jsonify(usr.get_api_data())

        id_, params = _build_data(self.apikey, 'update_user',

                                  userid=TEST_USER_ADMIN_LOGIN)

            def __exit__(self, exc_type, exc_val, exc_tb):

                anon = User.get_default_user()

                anon.active = not status

                Session().add(anon)

                Session().commit()

        return context()

    def _get_repo_create_params(self, **custom):

        defs = dict(

            repo_name=None,

            repo_type='hg',

            clone_uri='',

            repo_group='-1',

            repo_description='DESC',

            repo_private=False,

            repo_landing_rev='rev:tip',

            repo_copy_permissions=False,

            repo_state=Repository.STATE_CREATED,

        )

        defs.update(custom)

        if 'repo_name_full' not in custom:

            defs.update({'repo_name_full': defs['repo_name']})

        # fix the repo name if passed as repo_name_full

        if defs['repo_name']:

            defs['repo_name'] = defs['repo_name'].split('/')[-1]

        return defs

    def _get_group_create_params(self, **custom):

        defs = dict(

            group_name=None,

            group_description='DESC',

            group_parent_id=None,

            perms_updates=[],

            perms_new=[],

            enable_locking=False,

            recursive=False

        )

        defs.update(custom)

        return defs

    def _get_user_create_params(self, name, **custom):

        defs = dict(

            username=name,

            password='qweqwe',

            firstname='TestUser',

            lastname='Test',

            active=True,

            admin=False,

            extern_type='rhodecode',

            extern_name=None

        )

        defs.update(custom)

        return defs

    def _get_user_group_create_params(self, name, **custom):

        defs = dict(

            users_group_name=name,

            user_group_description='DESC',

            users_group_active=True,

            user_group_data={},

        )

        defs.update(custom)

        return defs

    def create_repo(self, name, **kwargs):

        if 'skip_if_exists' in kwargs:

            del kwargs['skip_if_exists']

            r = Repository.get_by_repo_name(name)

            if r:

                return r

        if isinstance(kwargs.get('repo_group'), RepoGroup):

            kwargs['repo_group'] = kwargs['repo_group'].group_id

        form_data = self._get_repo_create_params(repo_name=name, **kwargs)

        cur_user = kwargs.get('cur_user', TEST_USER_ADMIN_LOGIN)

        RepoModel().create(form_data, cur_user)

        Session().commit()

        return Repository.get_by_repo_name(name)

    def create_fork(self, repo_to_fork, fork_name, **kwargs):

        repo_to_fork = Repository.get_by_repo_name(repo_to_fork)

        form_data = self._get_repo_create_params(repo_name=fork_name,

                                            fork_parent_id=repo_to_fork,

                                            repo_type=repo_to_fork.repo_type,

                                            **kwargs)

        form_data['update_after_clone'] = False

        #TODO: fix it !!

from rhodecode.tests import *

from rhodecode.model.db import Notification, User

from rhodecode.model.user import UserModel

from rhodecode.model.notification import NotificationModel

from rhodecode.model.meta import Session

class TestNotificationsController(TestController):

    def tearDown(self):

        for n in Notification.query().all():

            inst = Notification.get(n.notification_id)

            Session().delete(inst)

        Session().commit()

    def test_index(self):

        self.log_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                          firstname='u1', lastname='u1')

        u1 = u1.user_id

        response = self.app.get(url('notifications'))

        response.mustcontain('<div class="table">No notifications here yet</div>')

        cur_user = self._get_logged_user()

        notif = NotificationModel().create(created_by=u1, subject=u'test_notification_1',

                                           body=u'notification_1', recipients=[cur_user])

        Session().commit()

        response = self.app.get(url('notifications'))

        response.mustcontain('id="notification_%s"' % notif.notification_id)

    def test_delete(self):

        self.log_user()

        cur_user = self._get_logged_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                               firstname='u1', lastname='u1')

        u2 = UserModel().create_or_update(username='u2', password='qweqwe',

                                               firstname='u2', lastname='u2')

        # make notifications

        notification = NotificationModel().create(created_by=cur_user,

                                                  subject=u'test',

                                                  body=u'hi there',

                                                  recipients=[cur_user, u1, u2])

        Session().commit()

        u1 = User.get(u1.user_id)

        u2 = User.get(u2.user_id)

        # check DB

        get_notif = lambda un: [x.notification for x in un]

        self.assertEqual(get_notif(cur_user.notifications), [notification])

        self.assertEqual(get_notif(u1.notifications), [notification])

        self.assertEqual(get_notif(u2.notifications), [notification])

        cur_usr_id = cur_user.user_id

        response = self.app.delete(url('notification',

                                       notification_id=

                                       notification.notification_id))

        self.assertEqual(response.body, 'ok')

        cur_user = User.get(cur_usr_id)

        self.assertEqual(cur_user.notifications, [])

    def test_show(self):

        self.log_user()

        cur_user = self._get_logged_user()

        u1 = UserModel().create_or_update(username='u1', password='qweqwe',

                                          firstname='u1', lastname='u1')

        u2 = UserModel().create_or_update(username='u2', password='qweqwe',

                                          firstname='u2', lastname='u2')

        subject = u'test'

        notif_body = u'hi there'

        notification = NotificationModel().create(created_by=cur_user,

                                                  subject=subject,

                                                  body=notif_body,

                                                  recipients=[cur_user, u1, u2])

        response = self.app.get(url('notification',

                                    notification_id=notification.notification_id))

        response.mustcontain(subject)

        response.mustcontain(notif_body)

    def test_create_err(self):

        self.log_user()

        username = 'new_user'

        password = ''

        name = 'name'

        lastname = 'lastname'

        email = 'errmail.com'

        response = self.app.post(url('users'), {'username': username,

                                               'password': password,

                                               'name': name,

                                               'active': False,

                                               'lastname': lastname,

                                               'email': email})

        msg = validators.ValidUsername(False, {})._messages['system_invalid_username']

        msg = h.html_escape(msg % {'username': 'new_user'})

        response.mustcontain("""<span class="error-message">%s</span>""" % msg)

        response.mustcontain("""<span class="error-message">Please enter a value</span>""")

        response.mustcontain("""<span class="error-message">An email address must contain a single @</span>""")

        def get_user():

            Session().query(User).filter(User.username == username).one()

        self.assertRaises(NoResultFound, get_user), 'found user in database'

    def test_new(self):

        self.log_user()

        response = self.app.get(url('new_user'))

    @parameterized.expand(

        [('firstname', {'firstname': 'new_username'}),

         ('lastname', {'lastname': 'new_username'}),

         ('admin', {'admin': True}),

         ('admin', {'admin': False}),

         ('extern_type', {'extern_type': 'ldap'}),

         ('extern_type', {'extern_type': None}),

         ('extern_name', {'extern_name': 'test'}),

         ('extern_name', {'extern_name': None}),

         ('active', {'active': False}),

         ('active', {'active': True}),

         ('email', {'email': 'some@email.com'}),

        # ('new_password', {'new_password': 'foobar123',

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_update(self, name, attrs):

        self.log_user()

        usr = fixture.create_user(self.test_user_1, password='qweqwe',

                                  extern_type='rhodecode',

                                  extern_name=self.test_user_1,

                                  skip_if_exists=True)

        Session().commit()

        params = usr.get_api_data()

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        params.update(attrs)

        if name == 'email':

            params['emails'] = [attrs['email']]

        if name == 'extern_type':

            #cannot update this via form, expected value is original one

            params['extern_type'] = "rhodecode"

        if name == 'extern_name':

            #cannot update this via form, expected value is original one

            params['extern_name'] = self.test_user_1

            # special case since this user is not

                                          # logged in yet his data is not filled

                                          # so we use creation data

        response = self.app.put(url('user', id=usr.user_id), params)

        self.checkSessionFlash(response, 'User updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data()

        updated_params.update({'password_confirmation': ''})

        updated_params.update({'new_password': ''})

        self.assertEqual(params, updated_params)

    def test_delete(self):

        self.log_user()

        username = 'newtestuserdeleteme'

        fixture.create_user(name=username)

        new_user = Session().query(User)\

            .filter(User.username == username).one()

        response = self.app.delete(url('user', id=new_user.user_id))

        self.checkSessionFlash(response, 'Successfully deleted user')

    def test_show(self):

        response = self.app.get(url('user', id=1))

    def test_edit(self):

        self.log_user()

        user = User.get_by_username(TEST_USER_ADMIN_LOGIN)

# -*- coding: utf-8 -*-

from rhodecode.tests import *

from rhodecode.tests.fixture import Fixture

from rhodecode.model.db import Repository

from rhodecode.model.repo import RepoModel

from rhodecode.model.user import UserModel

from rhodecode.model.meta import Session

fixture = Fixture()

from rhodecode.tests import *

class _BaseTest(TestController):

    """

    Write all tests here

    """

    REPO = None

    REPO_TYPE = None

    NEW_REPO = None

    REPO_FORK = None

    @classmethod

    def setup_class(cls):

        pass

    @classmethod

    def teardown_class(cls):

        pass

    def setUp(self):

        self.username = u'forkuser'

        self.password = u'qweqwe'

        self.u1 = fixture.create_user(self.username, password=self.password,

        Session().commit()

    def tearDown(self):

        Session().delete(self.u1)

        Session().commit()

    def test_index(self):

        self.log_user()

        repo_name = self.REPO

        response = self.app.get(url(controller='forks', action='forks',

                                    repo_name=repo_name))

        response.mustcontain("""There are no forks yet""")

    def test_no_permissions_to_fork(self):

        usr = self.log_user(TEST_USER_REGULAR_LOGIN,

                            TEST_USER_REGULAR_PASS)['user_id']

        user_model = UserModel()

        user_model.revoke_perm(usr, 'hg.fork.repository')

        user_model.grant_perm(usr, 'hg.fork.none')

        u = UserModel().get(usr)

        u.inherit_default_permissions = False

        Session().commit()

        # try create a fork

        repo_name = self.REPO

        self.app.post(url(controller='forks', action='fork_create',

                          repo_name=repo_name), {}, status=403)

    def test_index_with_fork(self):

        self.log_user()

        # create a fork

        fork_name = self.REPO_FORK

        description = 'fork of vcs test'

        repo_name = self.REPO

        org_repo = Repository.get_by_repo_name(repo_name)

        creation_args = {

            'repo_name': fork_name,

            'repo_group': '',

            'fork_parent_id': org_repo.repo_id,

            'repo_type': self.REPO_TYPE,

            'description': description,

            'private': 'False',

            'landing_rev': 'rev:tip'}

        self.app.post(url(controller='forks', action='fork_create',

                          repo_name=repo_name), creation_args)

    def tearDown(self):

        for n in Notification.query().all():

            Session().delete(n)

        Session().commit()

        self.assertEqual(Notification.query().all(), [])

    def test_index(self):

        response = self.app.get(url(controller='login', action='index'))

        self.assertEqual(response.status, '200 OK')

        # Test response...

    def test_login_admin_ok(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'test_admin',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response.session['rhodecode_user'].get('username'),

                         'test_admin')

        response = response.follow()

        response.mustcontain('/%s' % HG_REPO)

    def test_login_regular_ok(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'test_regular',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response.session['rhodecode_user'].get('username'),

                         'test_regular')

        response = response.follow()

        response.mustcontain('/%s' % HG_REPO)

    def test_login_ok_came_from(self):

        test_came_from = '/_admin/users'

        response = self.app.post(url(controller='login', action='index',

                                     came_from=test_came_from),

                                 {'username': 'test_admin',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('Users administration')

    @parameterized.expand([

          ('data:text/html,<script>window.alert("xss")</script>',),

          ('file:///etc/passwd',),

          ('ftp://some.ftp.server',),

          ('http://other.domain',),

    ])

    def test_login_bad_came_froms(self, url_came_from):

        response = self.app.post(url(controller='login', action='index',

                                     came_from=url_came_from),

                                 {'username': 'test_admin',

                                  'password': 'test12'})

        self.assertEqual(response.status, '302 Found')

        self.assertEqual(response._environ['paste.testing_variables']

                         ['tmpl_context'].came_from, '/')

        response = response.follow()

        self.assertEqual(response.status, '200 OK')

    def test_login_short_password(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'test_admin',

                                  'password': 'as'})

        self.assertEqual(response.status, '200 OK')

        response.mustcontain('Enter 3 characters or more')

    def test_login_wrong_username_password(self):

        response = self.app.post(url(controller='login', action='index'),

                                 {'username': 'error',

                                  'password': 'test12'})

        response.mustcontain('invalid user name')

        response.mustcontain('invalid password')

    #==========================================================================

    # REGISTRATIONS

    #==========================================================================

    def test_register(self):

        response = self.app.get(url(controller='login', action='register'))

        response.mustcontain('Sign Up')

    def test_register_err_same_username(self):

        uname = 'test_admin'

        response = self.app.post(url(controller='login', action='register'),

                                            {'username': uname,

                                             'password': 'test12',

                                             'password_confirmation': 'test12',

                                             'email': 'goodmail@domain.com',

                                             'firstname': 'test',

                                             'lastname': 'test'})

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

        response = self.app.post(url('my_account_emails'),)

        self.checkSessionFlash(response, 'Please enter an email address')

    def test_my_account_my_emails_add_remove(self):

        self.log_user()

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

        response = self.app.post(url('my_account_emails'),

                                 {'new_email': 'foo@barz.com'})

        response = self.app.get(url('my_account_emails'))

        from rhodecode.model.db import UserEmailMap

        email_id = UserEmailMap.query()\

            .filter(UserEmailMap.user == User.get_by_username(TEST_USER_ADMIN_LOGIN))\

            .filter(UserEmailMap.email == 'foo@barz.com').one().email_id

        response.mustcontain('foo@barz.com')

        response.mustcontain('<input id="del_email_id" name="del_email_id" type="hidden" value="%s" />' % email_id)

        response = self.app.post(url('my_account_emails'),

                                 {'del_email_id': email_id, '_method': 'delete'})

        self.checkSessionFlash(response, 'Removed email from user')

        response = self.app.get(url('my_account_emails'))

        response.mustcontain('No additional emails specified')

    @parameterized.expand(

        [('firstname', {'firstname': 'new_username'}),

         ('lastname', {'lastname': 'new_username'}),

         ('admin', {'admin': True}),

         ('admin', {'admin': False}),

         ('extern_type', {'extern_type': 'ldap'}),

         ('extern_type', {'extern_type': None}),

         #('extern_name', {'extern_name': 'test'}),

         #('extern_name', {'extern_name': None}),

         ('active', {'active': False}),

         ('active', {'active': True}),

         ('email', {'email': 'some@email.com'}),

        # ('new_password', {'new_password': 'foobar123',

        #                   'password_confirmation': 'foobar123'})

        ])

    def test_my_account_update(self, name, attrs):

        usr = fixture.create_user(self.test_user_1, password='qweqwe',

                                  extern_type='rhodecode',

                                  extern_name=self.test_user_1,

                                  skip_if_exists=True)

        params = usr.get_api_data()  # current user data

        user_id = usr.user_id

        self.log_user(username=self.test_user_1, password='qweqwe')

        params.update({'password_confirmation': ''})

        params.update({'new_password': ''})

        params.update({'extern_type': 'rhodecode'})

        params.update({'extern_name': self.test_user_1})

        params.update(attrs)

        response = self.app.post(url('my_account'), params)

        self.checkSessionFlash(response,

                               'Your account was updated successfully')

        updated_user = User.get_by_username(self.test_user_1)

        updated_params = updated_user.get_api_data()

        updated_params.update({'password_confirmation': ''})

        updated_params.update({'new_password': ''})

        params['last_login'] = updated_params['last_login']

        if name == 'email':

            params['emails'] = [attrs['email']]

        if name == 'extern_type':

            #cannot update this via form, expected value is original one

            params['extern_type'] = "rhodecode"

        if name == 'extern_name':

            #cannot update this via form, expected value is original one

            params['extern_name'] = str(user_id)

        if name == 'active':

            #my account cannot deactivate account

            params['active'] = True

        if name == 'admin':

            #my account cannot make you an admin !

            params['admin'] = False

        self.assertEqual(params, updated_params)

    def test_my_account_update_err_email_exists(self):

        self.log_user()

        new_email = 'test_regular@mail.com'  # already exisitn email

        response = self.app.post(url('my_account'),

                                params=dict(

                                    username='test_admin',

from rhodecode.lib.auth import AuthUser

from rhodecode.model.meta import Session

fixture = Fixture()

def _destroy_project_tree(test_u1_id):

    Session.remove()

    repo_group = RepoGroup.get_by_group_name(group_name='g0')

    for el in reversed(repo_group.recursive_groups_and_repos()):

        if isinstance(el, Repository):

            RepoModel().delete(el)

        elif isinstance(el, RepoGroup):

            RepoGroupModel().delete(el, force_delete=True)

    u = User.get(test_u1_id)

    Session().delete(u)

    Session().commit()

def _create_project_tree():

    """

    Creates a tree of groups and repositories to test permissions

    structure

     [g0] - group `g0` with 3 subgroups

     |

     |__[g0_1] group g0_1 with 2 groups 0 repos

     |  |

     |  |__[g0_1_1] group g0_1_1 with 1 group 2 repos

     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r1>

     |  |   |__<g0/g0_1/g0_1_1/g0_1_1_r2>

     |  |__<g0/g0_1/g0_1_r1>

     |

     |__[g0_2] 2 repos

     |  |

     |  |__<g0/g0_2/g0_2_r1>

     |  |__<g0/g0_2/g0_2_r2>

     |

     |__[g0_3] 1 repo

        |

        |_<g0/g0_3/g0_3_r1>

        |_<g0/g0_3/g0_3_r2_private>