test_pipeline.py

import os
import pytest

@pytest.mark.parametrize('data', ['cora', 'citeseer', 'pubmed', 'csv', 'reddit',
                                  'co-buy-computer', 'ogbn-arxiv', 'ogbn-products'])
def test_nodepred_data(data):
    os.system(f'dgl configure nodepred --data {data} --model gcn')
    assert os.path.exists(f'nodepred_{data}_gcn.yaml')

    custom_cfg = f'custom_{data}_gcn.yaml'
    os.system(f'dgl configure nodepred --data {data} --model gcn --cfg {custom_cfg}')
    assert os.path.exists(custom_cfg)

    custom_script = f'{data}_gcn.py'
    os.system(f'dgl export --cfg {custom_cfg} --output {custom_script}')
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('model', ['gcn', 'gat', 'sage', 'sgc', 'gin'])
def test_nodepred_model(model):
    os.system(f'dgl configure nodepred --data cora --model {model}')
    assert os.path.exists(f'nodepred_cora_{model}.yaml')

    custom_cfg = f'custom_cora_{model}.yaml'
    os.system(f'dgl configure nodepred --data cora --model {model} --cfg {custom_cfg}')
    assert os.path.exists(custom_cfg)

    custom_script = f'cora_{model}.py'
    os.system(f'dgl export --cfg {custom_cfg} --output {custom_script}')
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('data', ['cora', 'citeseer', 'pubmed', 'csv', 'reddit',
                                  'co-buy-computer', 'ogbn-arxiv', 'ogbn-products'])
def test_nodepred_ns_data(data):
    os.system(f'dgl configure nodepred-ns --data {data} --model gcn')
    assert os.path.exists(f'nodepred-ns_{data}_gcn.yaml')

    custom_cfg = f'ns-custom_{data}_gcn.yaml'
    os.system(f'dgl configure nodepred-ns --data {data} --model gcn --cfg {custom_cfg}')
    assert os.path.exists(custom_cfg)

    custom_script = f'ns-{data}_gcn.py'
    os.system(f'dgl export --cfg {custom_cfg} --output {custom_script}')
    assert os.path.exists(custom_script)


@pytest.mark.parametrize('model', ['gcn', 'gat', 'sage'])
def test_nodepred_ns_model(model):
    os.system(f'dgl configure nodepred-ns --data cora --model {model}')
    assert os.path.exists(f'nodepred-ns_cora_{model}.yaml')

    custom_cfg = f'ns-custom_cora_{model}.yaml'
    os.system(f'dgl configure nodepred-ns --data cora --model {model} --cfg {custom_cfg}')
    assert os.path.exists(custom_cfg)

    custom_script = f'ns-cora_{model}.py'
    os.system(f'dgl export --cfg {custom_cfg} --output {custom_script}')
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('data', ['cora', 'citeseer', 'pubmed', 'csv', 'reddit',
                                  'co-buy-computer', 'ogbn-arxiv', 'ogbn-products', 'ogbl-collab',
                                  'ogbl-citation2'])
def test_linkpred_data(data):
    node_model = 'gcn'
    edge_model = 'ele'
    neg_sampler = 'global'
    custom_cfg = '_'.join([data, node_model, edge_model, neg_sampler]) + '.yaml'
    os.system('dgl configure linkpred --data {} --node-model {} --edge-model {} --neg-sampler {} --cfg {}'.format(
        data, node_model, edge_model, neg_sampler, custom_cfg))
    assert os.path.exists(custom_cfg)

    custom_script = '_'.join([data, node_model, edge_model, neg_sampler]) + '.py'
    os.system('dgl export --cfg {} --output {}'.format(custom_cfg, custom_script))
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('node_model', ['gcn' ,'gat', 'sage', 'sgc', 'gin'])
def test_linkpred_node_model(node_model):
    data = 'cora'
    edge_model = 'ele'
    neg_sampler = 'global'
    custom_cfg = '_'.join([data, node_model, edge_model, neg_sampler]) + '.yaml'
    os.system('dgl configure linkpred --data {} --node-model {} --edge-model {} --neg-sampler {} --cfg {}'.format(
        data, node_model, edge_model, neg_sampler, custom_cfg))
    assert os.path.exists(custom_cfg)

    custom_script = '_'.join([data, node_model, edge_model, neg_sampler]) + '.py'
    os.system('dgl export --cfg {} --output {}'.format(custom_cfg, custom_script))
    assert os.path.exists(custom_script)


@pytest.mark.parametrize('edge_model', ['ele', 'bilinear'])
def test_linkpred_edge_model(edge_model):
    data = 'cora'
    node_model = 'gcn'
    neg_sampler = 'global'
    custom_cfg = '_'.join([data, node_model, edge_model, neg_sampler]) + '.yaml'
    os.system('dgl configure linkpred --data {} --node-model {} --edge-model {} --neg-sampler {} --cfg {}'.format(
        data, node_model, edge_model, neg_sampler, custom_cfg))
    assert os.path.exists(custom_cfg)

    custom_script = '_'.join([data, node_model, edge_model, neg_sampler]) + '.py'
    os.system('dgl export --cfg {} --output {}'.format(custom_cfg, custom_script))
    assert os.path.exists(custom_script)


@pytest.mark.parametrize('neg_sampler', ['global', 'persource', ''])
def test_linkpred_neg_sampler(neg_sampler):
    data = 'cora'
    node_model = 'gcn'
    edge_model = 'ele'
    custom_cfg = f'{data}_{node_model}_{edge_model}_{neg_sampler}.yaml'
    if neg_sampler == '':
        os.system('dgl configure linkpred --data {} --node-model {} --edge-model {} --cfg {}'.format(
            data, node_model, edge_model, custom_cfg))
    else:
        os.system('dgl configure linkpred --data {} --node-model {} --edge-model {} --neg-sampler {} --cfg {}'.format(
            data, node_model, edge_model, neg_sampler, custom_cfg))
    assert os.path.exists(custom_cfg)

    custom_script = f'{data}_{node_model}_{edge_model}_{neg_sampler}.py'
    os.system('dgl export --cfg {} --output {}'.format(custom_cfg, custom_script))
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('data', ['csv', 'ogbg-molhiv', 'ogbg-molpcba'])
@pytest.mark.parametrize('model', ['gin', 'pna'])
def test_graphpred(data, model):
    os.system('dgl configure graphpred --data {} --model {}'.format(data, model))
    assert os.path.exists('graphpred_{}_{}.yaml'.format(data, model))

    custom_cfg = 'custom_{}_{}.yaml'.format(data, model)
    os.system('dgl configure graphpred --data {} --model {} --cfg {}'.format(data, model,
                                                                             custom_cfg))
    assert os.path.exists(custom_cfg)

    custom_script = '_'.join([data, model]) + '.py'
    os.system('dgl export --cfg {} --output {}'.format(custom_cfg, custom_script))
    assert os.path.exists(custom_script)

@pytest.mark.parametrize('recipe',
                         ['graphpred_hiv_gin.yaml',
                          'graphpred_hiv_pna.yaml',
                          'graphpred_pcba_gin.yaml',
                          'linkpred_cora_sage.yaml',
                          'linkpred_citation2_sage.yaml',
                          'linkpred_collab_sage.yaml',
                          'nodepred_citeseer_gat.yaml',
                          'nodepred_citeseer_gcn.yaml',
                          'nodepred_citeseer_sage.yaml',
                          'nodepred_cora_gat.yaml',
                          'nodepred_cora_gcn.yaml',
                          'nodepred_cora_sage.yaml',
                          'nodepred_pubmed_gat.yaml',
                          'nodepred_pubmed_gcn.yaml',
                          'nodepred_pubmed_sage.yaml',
                          'nodepred-ns_arxiv_gcn.yaml',
                          'nodepred-ns_product_sage.yaml'])
def test_recipe(recipe):
    # Remove all generated yaml files
    current_dir = os.listdir("./")
    for item in current_dir:
        if item.endswith(".yaml"):
            os.remove(item)

    os.system('dgl recipe get {}'.format(recipe))
    assert os.path.exists(recipe)

def test_node_cora():
    os.system('dgl configure nodepred --data cora --model gcn')
    os.system('dgl train --cfg nodepred_cora_gcn.yaml')
    assert os.path.exists('results')
    assert os.path.exists('results/run_0.pth')
    os.system('dgl configure-apply nodepred --cpt results/run_0.pth')
    assert os.path.exists('apply_nodepred_cora_gcn.yaml')
    os.system('dgl configure-apply nodepred --data cora --cpt results/run_0.pth --cfg apply.yaml')
    assert os.path.exists('apply.yaml')
    os.system('dgl apply --cfg apply.yaml')
    assert os.path.exists('apply_results/output.csv')
    os.system('dgl export --cfg apply.yaml --output apply.py')
    assert os.path.exists('apply.py')